# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: AGPL-3.0
"""
VikingFS: OpenViking file system abstraction layer

Encapsulates the AGFS binding client, providing file operation interface based on Viking URI.
Responsibilities:
- URI conversion (viking:// <-> /local/)
- L0/L1 reading (.abstract.md, .overview.md)
- Relation management (.relations.json)
- Semantic search (vector retrieval + rerank)
- Vector sync (sync vector store on rm/mv)
"""

import asyncio
import contextvars
import hashlib
import json
import os
import re
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import PurePath
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional, TypeVar, Union

from openviking.core.context import ContextLevel
from openviking.core.namespace import (
    canonicalize_uri,
    is_hidden_by_actor_peer_view,
    may_include_hidden_actor_peers,
)
from openviking.core.namespace import (
    is_accessible as namespace_is_accessible,
)
from openviking.core.retrieval_targets import resolve_retrieval_targets
from openviking.pyagfs import AsyncAGFSClient
from openviking.pyagfs.exceptions import (
    AGFSClientError,
    AGFSDirectoryNotEmptyError,
    AGFSHTTPError,
    AGFSInvalidOperationError,
    AGFSNotSupportedError,
)
from openviking.resource.watch_storage import is_watch_task_control_uri
from openviking.server.error_mapping import is_not_found_error, map_exception
from openviking.server.identity import RequestContext, Role
from openviking.storage.expr import And, PathScope, RawDSL
from openviking.storage.internal_names import (
    MULTIWRITE_PATH_LOCK_FILE,
    STORAGE_INTERNAL_ENTRY_NAMES,
)
from openviking.telemetry import get_current_telemetry
from openviking.utils.time_utils import format_iso8601, get_current_timestamp, parse_iso_datetime
from openviking_cli.exceptions import (
    FailedPreconditionError,
    InvalidArgumentError,
    NotFoundError,
    PermissionDeniedError,
)
from openviking_cli.session.user_id import UserIdentifier
from openviking_cli.utils.config.grep_config import GrepEngine
from openviking_cli.utils.logger import get_logger
from openviking_cli.utils.uri import VikingURI

if TYPE_CHECKING:
    from openviking.storage.transaction.lock_handle import LockHandle
    from openviking.storage.viking_vector_index_backend import VikingVectorIndexBackend
    from openviking_cli.utils.config import GrepConfig, RerankConfig, RetrievalConfig

logger = get_logger(__name__)

# Sentinel node_limit for internal callers that MUST enumerate an entire
# directory. ``ls()`` defaults to ``node_limit=1000`` to protect agent-facing
# context from being flooded, but internal system operations (parse merge,
# temp->final sync, summary DAG, vectorization) must see every child or they
# silently drop entries beyond the cap — e.g. a >1000-doc directory ingest only
# materializes its first 1000 subdirectories. Pass this explicitly at those
# call sites.
LS_ALL_NODES = 2**31 - 1
_T = TypeVar("_T")


def _ensure_non_empty_search_query(query: str) -> None:
    if not query.strip():
        raise InvalidArgumentError("Search query must not be empty.")


def _is_directory_not_empty_error(message: str) -> bool:
    """Check if an error message indicates a directory not empty error.

    Handles multiple possible error message formats from different backends.
    """
    msg = message.lower()
    return any(
        pattern in msg
        for pattern in [
            "directory not empty",
            "dir not empty",
            "directory is not empty",
        ]
    )


def _get_cpu_count() -> int:
    """Return the number of CPUs available to this process.

    Tries process_cpu_count (Python 3.13+, cgroup-aware),
    falls back to sched_getaffinity (Linux),
    then os.cpu_count (may report host CPUs in containers).
    """
    if hasattr(os, "process_cpu_count"):
        return os.process_cpu_count() or 1
    try:
        return len(os.sched_getaffinity(0))
    except (AttributeError, NotImplementedError):
        return os.cpu_count() or 1


def _get_abstract_worker_count() -> int:
    default = max(4, min(12, min(32, _get_cpu_count() + 4) // 2))
    env_val = os.getenv("OPENVIKING_FILE_OPS_CONCURRENCY")
    if env_val is not None:
        try:
            return max(1, int(env_val))
        except ValueError:
            pass
    return max(1, default)


_ABSTRACT_WORKER_COUNT = _get_abstract_worker_count()
_DEFAULT_GREP_FILE_CONCURRENCY = 32


# ========== Dataclass ==========


@dataclass
class RelationEntry:
    """Relation table entry."""

    id: str
    uris: List[str]
    reason: str = ""
    created_at: str = field(default_factory=get_current_timestamp)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "uris": self.uris,
            "reason": self.reason,
            "created_at": self.created_at,
        }

    @staticmethod
    def from_dict(data: Dict[str, Any]) -> "RelationEntry":
        return RelationEntry(**data)


# ========== Singleton Pattern ==========

_instance: Optional["VikingFS"] = None


def init_viking_fs(
    agfs: Any,
    query_embedder: Optional[Any] = None,
    rerank_config: Optional["RerankConfig"] = None,
    vector_store: Optional["VikingVectorIndexBackend"] = None,
    retrieval_config: Optional["RetrievalConfig"] = None,
    grep_config: Optional["GrepConfig"] = None,
    timeout: int = 10,
    enable_recorder: bool = False,
    encryptor: Optional[Any] = None,
) -> "VikingFS":
    """Initialize VikingFS singleton.

    Args:
        agfs: Pre-initialized AGFS client (HTTP or Binding)
        query_embedder: Embedder instance
        rerank_config: Rerank configuration
        retrieval_config: Retrieval ranking configuration
        grep_config: Grep engine configuration
        vector_store: Vector store instance
        enable_recorder: Whether to enable IO recording
        encryptor: FileEncryptor instance for encryption/decryption
    """
    global _instance

    _instance = VikingFS(
        agfs=agfs,
        query_embedder=query_embedder,
        rerank_config=rerank_config,
        vector_store=vector_store,
        retrieval_config=retrieval_config,
        grep_config=grep_config,
        encryptor=encryptor,
    )

    if enable_recorder:
        _enable_viking_fs_recorder(_instance)

    return _instance


def _enable_viking_fs_recorder(viking_fs: "VikingFS") -> None:
    """
    Enable recorder for a VikingFS instance.

    This wraps the VikingFS instance with recording capabilities.
    Called automatically when enable_recorder=True in init_viking_fs.

    Args:
        viking_fs: VikingFS instance to enable recording for
    """
    from openviking.eval.recorder import RecordingVikingFS, get_recorder

    recorder = get_recorder()
    if not recorder.enabled:
        from openviking.eval.recorder import init_recorder

        init_recorder(enabled=True)

    global _instance
    _instance = RecordingVikingFS(viking_fs)
    logger.info("[VikingFS] IO Recorder enabled")


def enable_viking_fs_recorder() -> None:
    """
    Enable recorder for the global VikingFS singleton.

    This function wraps the existing VikingFS's AGFS client with recording.
    Must be called after init_viking_fs().
    """
    global _instance
    if _instance is None:
        raise RuntimeError("VikingFS not initialized. Call init_viking_fs() first.")
    _enable_viking_fs_recorder(_instance)


def get_viking_fs() -> "VikingFS":
    """Get VikingFS singleton."""
    if _instance is None:
        raise RuntimeError("VikingFS not initialized. Call init_viking_fs() first.")
    return _instance


# ========== VikingFS Main Class ==========


class VikingFS:
    """RAGFS-based OpenViking file system.

    APIs are divided into two categories:
    - RAGFS basic commands (direct forwarding): read, ls, write, mkdir, rm, mv, grep, stat
    - VikingFS specific capabilities: abstract, overview, find, search, relations, link, unlink

    Uses Rust binding mode: Use RAGFSBindingClient to directly use RAGFS implementation
    """

    def __init__(
        self,
        agfs: Any,
        query_embedder: Optional[Any] = None,
        rerank_config: Optional["RerankConfig"] = None,
        vector_store: Optional["VikingVectorIndexBackend"] = None,
        retrieval_config: Optional["RetrievalConfig"] = None,
        grep_config: Optional["GrepConfig"] = None,
        timeout: int = 10,
        encryptor: Optional[Any] = None,
    ):
        self.agfs = agfs
        self._async_agfs = AsyncAGFSClient(agfs)
        self.query_embedder = query_embedder
        self.rerank_config = rerank_config
        self.vector_store = vector_store
        self.retrieval_config = retrieval_config
        self.grep_config = grep_config
        self._encryptor = encryptor
        self._count_cache: Dict[str, tuple] = {}  # cache_key → (count, timestamp)
        self._count_cache_max_size = 1024
        self._fulltext_available: Optional[bool] = None  # cached result of _collection_has_fulltext
        self._bound_ctx: contextvars.ContextVar[Optional[RequestContext]] = contextvars.ContextVar(
            "vikingfs_bound_ctx", default=None
        )
        self._background_tasks: set = set()

    @staticmethod
    def _default_ctx() -> RequestContext:
        return RequestContext(user=UserIdentifier.the_default_user(), role=Role.ROOT)

    def _ctx_or_default(self, ctx: Optional[RequestContext]) -> RequestContext:
        if ctx is not None:
            return ctx
        bound = self._bound_ctx.get()
        return bound or self._default_ctx()

    @contextmanager
    def bind_request_context(self, ctx: RequestContext):
        """Temporarily bind ctx for legacy internal call paths without explicit ctx param."""
        token = self._bound_ctx.set(ctx)
        try:
            yield
        finally:
            self._bound_ctx.reset(token)

    @staticmethod
    def _normalize_uri(uri: str) -> str:
        """Normalize short-format URIs to the canonical viking:// form."""
        if uri.startswith("viking://"):
            return uri
        return VikingURI.normalize(uri)

    @classmethod
    def _normalized_uri_parts(cls, uri: str) -> tuple[str, List[str]]:
        """Normalize a URI and reject ambiguous or platform-specific path traversal forms."""
        normalized = cls._normalize_uri(uri)
        parts = [p for p in normalized[len("viking://") :].strip("/").split("/") if p]

        for part in parts:
            if part in {".", ".."}:
                raise PermissionDeniedError(
                    f"Unsafe URI traversal segment '{part}' in {normalized}",
                    resource=normalized,
                )
            if "\\" in part:
                raise PermissionDeniedError(
                    f"Unsafe URI path separator '\\\\' in component '{part}' of {normalized}",
                    resource=normalized,
                )
            if len(part) >= 2 and part[1] == ":" and part[0].isalpha():
                raise PermissionDeniedError(
                    f"Unsafe URI drive-prefixed component '{part}' in {normalized}",
                    resource=normalized,
                )

        return normalized, parts

    # TODO: Once pathlock moves down into ragfs, stop reconstructing the
    # encrypted mount-relative path in Python and derive the lock target from
    # the same backend-side source of truth.
    def _encrypted_mount_relative_path(self, path: str) -> tuple[str, str]:
        """Return the mount prefix and mount-relative path used by Rust encrypted writes."""
        normalized = path if path.startswith("/") else f"/{path}"
        parts = [part for part in normalized.split("/") if part]
        if len(parts) < 2:
            return "", normalized
        return f"/{parts[0]}", f"/{'/'.join(parts[1:])}"

    def _encrypted_temp_path(self, path: str) -> str:
        """Build the deterministic internal `.encrypt` temp-file path for a final file path."""
        mount_prefix, relative_path = self._encrypted_mount_relative_path(path)
        relative_parts = [part for part in relative_path.split("/") if part]
        if len(relative_parts) >= 2 and relative_parts[0] == "local":
            temp_root = f"/local/{relative_parts[1]}/temp/.encrypt_stage"
        elif len(relative_parts) >= 2:
            temp_root = f"/{relative_parts[0]}/temp/.encrypt_stage"
        else:
            temp_root = "/temp/.encrypt_stage"
        digest = hashlib.sha256(relative_path.encode("utf-8")).hexdigest()
        return f"{mount_prefix}{temp_root}/{digest}.encrypt"

    def _encrypted_write_lock_paths(self, path: str) -> List[str]:
        """Return the final and temp paths protected by the encrypted write protocol."""
        return [path, self._encrypted_temp_path(path)]

    async def _run_with_encrypted_write_lock(
        self,
        path: str,
        operation: Callable[[], Awaitable[_T]],
        lock_handle: Optional["LockHandle"] = None,
    ) -> _T:
        """Run a write operation under a dual-path exact lock when encryption is enabled."""
        if self._encryptor is None:
            return await operation()

        from openviking.storage.transaction import LockContext, get_lock_manager

        async with LockContext(
            get_lock_manager(),
            self._encrypted_write_lock_paths(path),
            lock_mode="exact",
            handle=lock_handle,
        ):
            return await operation()

    def _ensure_access(self, uri: str, ctx: Optional[RequestContext]) -> None:
        real_ctx = self._ctx_or_default(ctx)
        normalized_uri, _ = self._normalized_uri_parts(uri)
        if not self._is_accessible(normalized_uri, real_ctx):
            raise PermissionDeniedError(f"Access denied for {uri}", resource=normalized_uri)

    def _ensure_mutable_access(self, uri: str, ctx: Optional[RequestContext]) -> None:
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        normalized_uri, _ = self._normalized_uri_parts(uri)
        if is_hidden_by_actor_peer_view(normalized_uri, real_ctx) or may_include_hidden_actor_peers(
            normalized_uri, real_ctx
        ):
            raise PermissionDeniedError(f"Access denied for {uri}", resource=normalized_uri)
        self._ensure_supported_write_namespace(normalized_uri)
        if real_ctx.role != Role.ROOT and normalized_uri.rstrip("/") == "viking://temp":
            raise PermissionDeniedError(
                "Temp root is read-only for non-root users",
                resource=normalized_uri,
            )

    def _ensure_delete_access(self, uri: str, ctx: Optional[RequestContext]) -> None:
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        normalized_uri, _ = self._normalized_uri_parts(uri)
        if is_hidden_by_actor_peer_view(normalized_uri, real_ctx) or may_include_hidden_actor_peers(
            normalized_uri, real_ctx
        ):
            raise PermissionDeniedError(f"Access denied for {uri}", resource=normalized_uri)
        self._ensure_supported_delete_namespace(normalized_uri)
        if real_ctx.role != Role.ROOT and normalized_uri.rstrip("/") == "viking://temp":
            raise PermissionDeniedError(
                "Temp root is read-only for non-root users",
                resource=normalized_uri,
            )

    def _ensure_supported_delete_namespace(self, normalized_uri: str) -> None:
        parts = [p for p in normalized_uri[len("viking://") :].strip("/").split("/") if p]
        if not parts:
            raise PermissionDeniedError(
                "Deleting viking:// is not supported; use a concrete scope instead.",
                resource=normalized_uri,
            )
        if parts == ["user"]:
            raise PermissionDeniedError(
                "Deleting viking://user is not supported; use an explicit user namespace "
                "or current-user content path instead.",
                resource=normalized_uri,
            )
        if parts == ["agent"]:
            # Parity with _ensure_supported_write_namespace, which forbids the
            # account-shared viking://agent root: deleting it would recursively
            # wipe every account's agent skills/endpoints/tools/payments. Concrete
            # sub-paths (viking://agent/skills/...) remain deletable.
            raise PermissionDeniedError(
                "Deleting viking://agent root is not supported; use a concrete "
                "agent sub-path (e.g. viking://agent/skills/...) instead.",
                resource=normalized_uri,
            )

    def _ensure_supported_write_namespace(self, normalized_uri: str) -> None:
        parts = [p for p in normalized_uri[len("viking://") :].strip("/").split("/") if p]
        if parts == ["user"]:
            raise PermissionDeniedError(
                "Writing viking://user is not supported; use an explicit user namespace "
                "or current-user content path instead.",
                resource=normalized_uri,
            )
        if parts and parts[0] == "session":
            raise PermissionDeniedError(
                f"Writing {normalized_uri} is not supported; use user-owned namespaces instead.",
                resource=normalized_uri,
            )
        if parts and parts[0] == "agent":
            if len(parts) >= 2 and parts[1] not in {"skills", "endpoints", "tools", "payments"}:
                raise PermissionDeniedError(
                    "viking://agent/{agent_id} is deprecated. Use viking://user/.../peers/{agent_id} instead.",
                    resource=normalized_uri,
                )
            if len(parts) < 2:
                raise PermissionDeniedError(
                    "Writing to viking://agent root is not supported.",
                    resource=normalized_uri,
                )

    # ========== AGFS Basic Commands ==========

    async def read(
        self,
        uri: str,
        offset: int = 0,
        size: int = -1,
        ctx: Optional[RequestContext] = None,
    ) -> bytes:
        """Read file"""
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)

        # Decryption + offset/size slicing now happen inside the ragfs encryption layer
        # (when configured); the plaintext stack reads bytes directly. Either way, pass the
        # offset/size through and let the Rust layer return the requested slice.
        last_not_found: Optional[Exception] = None
        for path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, path, primary_path, real_ctx):
                continue
            try:
                result = await self._async_agfs.read(path, offset, size)
                break
            except Exception as exc:
                if is_not_found_error(exc):
                    last_not_found = exc
                    continue
                raise
        else:
            raise NotFoundError(uri, "file") from last_not_found
        if isinstance(result, bytes):
            raw = result
        elif result is not None and hasattr(result, "content"):
            raw = result.content
        else:
            raw = b""

        return raw

    async def write(
        self,
        uri: str,
        data: Union[bytes, str],
        ctx: Optional[RequestContext] = None,
    ) -> str:
        """Write file"""
        self._ensure_mutable_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        if isinstance(data, str):
            data = data.encode("utf-8")

        # Encryption (when configured) happens inside the ragfs layer keyed by account_id.
        return await self._async_agfs.write(path, data)

    async def mkdir(
        self,
        uri: str,
        mode: str = "755",
        exist_ok: bool = False,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Create directory."""
        self._ensure_mutable_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        # Always ensure parent directories exist before creating this directory
        await self._ensure_parent_dirs(path, ctx=ctx)
        try:
            await self._async_agfs.mkdir(path)
        except Exception as exc:
            message = str(exc).lower()
            already_exists = "exist" in message or "already" in message
            if exist_ok and already_exists:
                return

    async def rm(
        self,
        uri: str,
        recursive: bool = False,
        ctx: Optional[RequestContext] = None,
        lock_handle: Optional["LockHandle"] = None,
    ) -> Dict[str, Any]:
        """Delete file/directory + recursively update vector index.

        This method is idempotent: deleting a non-existent file succeeds
        after cleaning up any orphan index records.

        Acquires a path lock, deletes VectorDB records, then FS files.
        Raises ResourceBusyError when the target is locked by an ongoing
        operation (e.g. semantic processing).

        Returns:
            Dict with 'estimated_deleted_count' indicating the estimated number
            of nodes deleted from vector index.
        """
        from openviking.storage.errors import LockAcquisitionError, ResourceBusyError
        from openviking.storage.transaction import LockContext, get_lock_manager

        self._ensure_delete_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        target_uri = self._path_to_uri(path, ctx=ctx)

        async def _estimate_deleted_count(target_path: str, real_ctx: RequestContext) -> int:
            """Estimate number of nodes to be deleted using vector index."""
            vector_store = self._get_vector_store()
            if not vector_store:
                return 0
            try:
                target_canonical_uri = canonicalize_uri(
                    self._path_to_uri(target_path, ctx=real_ctx), real_ctx
                )
                filter_expr = PathScope("uri", target_canonical_uri, depth=-1)
                return await vector_store.count(filter=filter_expr, ctx=real_ctx)
            except Exception as e:
                logger.warning(f"[VikingFS] Failed to count nodes before delete: {e}")
                return 0

        # Check existence and determine lock strategy
        try:
            stat = await self._async_agfs.stat(path)
            is_dir = stat.get("isDir", False) if isinstance(stat, dict) else False
        except Exception as exc:
            if not is_not_found_error(exc):
                mapped = map_exception(exc, resource=uri)
                if mapped is not None:
                    raise mapped from exc
                raise
            # Path does not exist: clean up any orphan index records and return
            uris_to_delete = await self._collect_uris(path, recursive, ctx=ctx)
            uris_to_delete.append(target_uri)
            real_ctx = self._ctx_or_default(ctx)
            estimated_count = await _estimate_deleted_count(path, real_ctx)
            await self._delete_from_vector_store(uris_to_delete, ctx=ctx)
            logger.info(f"[VikingFS] rm target not found, cleaned orphan index: {uri}")
            return {"estimated_deleted_count": estimated_count}

        if is_dir:
            if not recursive:
                raise FailedPreconditionError(
                    f"Cannot remove directory without --recursive: {uri}",
                    details={"resource": uri, "expected_flag": "recursive"},
                )
            lock_paths = [path]
            lock_mode = "tree"
        else:
            lock_paths = [path]
            lock_mode = "exact"

        try:
            async with LockContext(
                get_lock_manager(),
                lock_paths,
                lock_mode=lock_mode,
                handle=lock_handle,
            ):
                uris_to_delete = await self._collect_uris(path, recursive, ctx=ctx)
                uris_to_delete.append(target_uri)
                real_ctx = self._ctx_or_default(ctx)
                estimated_count = await _estimate_deleted_count(path, real_ctx)
                await self._delete_from_vector_store(uris_to_delete, ctx=ctx)
                try:
                    result = await self._async_agfs.rm(path, recursive=recursive)
                except AGFSDirectoryNotEmptyError:
                    raise FailedPreconditionError(
                        f"Directory not empty: {uri}. Use recursive=True to delete non-empty directories."
                    )
                except RuntimeError as e:
                    # Fallback for older versions without typed exceptions
                    if _is_directory_not_empty_error(str(e)):
                        raise FailedPreconditionError(
                            f"Directory not empty: {uri}. Use recursive=True to delete non-empty directories."
                        )
                    raise
                # Add estimated_deleted_count to the result
                if isinstance(result, dict):
                    result["estimated_deleted_count"] = estimated_count
                else:
                    result = {"estimated_deleted_count": estimated_count}
                return result
        except LockAcquisitionError:
            raise ResourceBusyError(f"Resource is being processed: {uri}", uri=uri)

    async def mv(
        self,
        old_uri: str,
        new_uri: str,
        ctx: Optional[RequestContext] = None,
        lock_handle: Optional["LockHandle"] = None,
    ) -> Dict[str, Any]:
        """Move file/directory + recursively update vector index.

        Implemented as cp + rm to avoid lock files being carried by FS mv.
        On VectorDB update failure the copy is cleaned up so the source stays intact.
        """
        from openviking.storage.transaction import LockContext, get_lock_manager

        self._ensure_mutable_access(old_uri, ctx)
        # mv is implemented as copy + recursive rm of the source (see the
        # ``rm(old_path, recursive=is_dir)`` below), so the source must also clear
        # the delete guard. Without this, a protected account root such as
        # ``viking://`` — which rm() rejects up front (#2873) — could still be
        # destroyed via mv, since the write guard alone permits the bare root.
        self._ensure_delete_access(old_uri, ctx)
        self._ensure_mutable_access(new_uri, ctx)
        old_path = self._uri_to_path(old_uri, ctx=ctx)
        new_path = self._uri_to_path(new_uri, ctx=ctx)
        target_uri = self._path_to_uri(old_path, ctx=ctx)

        # Verify source exists and determine type before locking.
        try:
            stat = await self._async_agfs.stat(old_path)
            is_dir = stat.get("isDir", False) if isinstance(stat, dict) else False
        except Exception as exc:
            if not is_not_found_error(exc):
                mapped = map_exception(exc, resource=old_uri)
                if mapped is not None:
                    raise mapped from exc
                raise
            raise FileNotFoundError(f"mv source not found: {old_uri}") from exc

        if not is_dir:
            if new_uri.rstrip("/") != new_uri:
                raise InvalidArgumentError(
                    f"mv destination for a file must include the target file name: {new_uri}",
                    details={"from_uri": old_uri, "to_uri": new_uri},
                )
            try:
                destination_stat = await self._async_agfs.stat(new_path)
            except Exception as exc:
                if not is_not_found_error(exc):
                    mapped = map_exception(exc, resource=new_uri)
                    if mapped is not None:
                        raise mapped from exc
                    raise
            else:
                if isinstance(destination_stat, dict) and destination_stat.get("isDir", False):
                    raise InvalidArgumentError(
                        f"mv destination for a file must include the target file name: {new_uri}",
                        details={"from_uri": old_uri, "to_uri": new_uri},
                    )

        lock_context = (
            LockContext(
                get_lock_manager(),
                [old_path],
                lock_mode="mv",
                mv_dst_path=new_path,
                src_is_dir=True,
                handle=lock_handle,
            )
            if is_dir
            else LockContext(
                get_lock_manager(),
                [old_path, new_path],
                lock_mode="exact",
                handle=lock_handle,
            )
        )

        async with lock_context:
            uris_to_move = await self._collect_uris(old_path, recursive=True, ctx=ctx)
            uris_to_move.append(target_uri)

            # Check if it's temp directory (files already encrypted)
            is_temp = old_uri.startswith("viking://temp/")

            # Copy source to destination. Source must stay intact until vector updates succeed.
            try:
                await self._copy_for_mv(
                    old_uri=old_uri,
                    new_uri=new_uri,
                    old_path=old_path,
                    new_path=new_path,
                    is_dir=is_dir,
                    is_temp=is_temp,
                    ctx=ctx,
                )
            except Exception as e:
                if "not found" in str(e).lower():
                    await self._delete_from_vector_store(uris_to_move, ctx=ctx)
                    logger.info(f"[VikingFS] mv source not found, cleaned orphan index: {old_uri}")
                raise

            # Remove carried lock file from the copy (directory only)
            if is_dir:
                carried_lock = new_path.rstrip("/") + f"/{MULTIWRITE_PATH_LOCK_FILE}"
                try:
                    await self._async_agfs.rm(carried_lock)
                except Exception:
                    pass

            # Update VectorDB URIs (on failure, clean up the copy)
            try:
                await self._update_vector_store_uris(uris_to_move, old_uri, new_uri, ctx=ctx)
            except Exception:
                try:
                    if is_dir:
                        await self._async_agfs.rm(new_path, recursive=True)
                    else:
                        await self._async_agfs.rm(new_path)
                except Exception:
                    pass
                raise

            # Delete source
            await self._async_agfs.rm(old_path, recursive=is_dir)
            return {}

    async def system_sync_status(
        self, uri: str, ctx: Optional[RequestContext] = None
    ) -> Dict[str, Any]:
        """Return multi-write sync status for one Viking URI subtree."""
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        return await self._async_agfs.system_sync_status(
            path,
            fs_ctx={"account_id": real_ctx.account_id},
        )

    async def system_sync_retry(
        self, uri: str, ctx: Optional[RequestContext] = None
    ) -> Dict[str, Any]:
        """Retry multi-write sync for one Viking URI subtree."""
        self._ensure_mutable_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        return await self._async_agfs.system_sync_retry(
            path,
            fs_ctx={"account_id": real_ctx.account_id},
        )

    async def _copy_for_mv(
        self,
        old_uri: str,
        new_uri: str,
        old_path: str,
        new_path: str,
        is_dir: bool,
        is_temp: bool,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Copy source to destination for mv without deleting source."""
        if is_temp:
            await self._async_agfs.cp(
                old_path,
                new_path,
                recursive=is_dir,
                fs_ctx={"account_id": self._ctx_or_default(ctx).account_id},
            )
            return

        if is_dir:
            await self._copy_dir_through_vikingfs(old_uri, new_uri, ctx=ctx)
        else:
            await self._copy_file_through_vikingfs(old_uri, new_uri, ctx=ctx)

    async def _copy_dir_through_vikingfs(
        self,
        old_uri: str,
        new_uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Recursively copy a directory through VikingFS read/write hooks."""
        await self.mkdir(new_uri, exist_ok=True, ctx=ctx)

        entries = await self.ls(old_uri, show_all_hidden=True, ctx=ctx)
        for entry in entries:
            name = entry.get("name", "")
            if not name or name in (".", ".."):
                continue
            old_child_uri = f"{old_uri.rstrip('/')}/{name}"
            new_child_uri = f"{new_uri.rstrip('/')}/{name}"
            if entry.get("isDir"):
                await self._copy_dir_through_vikingfs(old_child_uri, new_child_uri, ctx=ctx)
            else:
                await self._copy_file_through_vikingfs(old_child_uri, new_child_uri, ctx=ctx)

    async def _copy_file_through_vikingfs(
        self,
        from_uri: str,
        to_uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Copy one file through VikingFS read/write hooks without deleting source."""
        content_bytes = await self.read_file_bytes(from_uri, ctx=ctx)
        await self.write_file_bytes(to_uri, content_bytes, ctx=ctx)

    async def grep(
        self,
        uri: str,
        pattern: str,
        exclude_uri: Optional[str] = None,
        case_insensitive: bool = False,
        node_limit: Optional[int] = None,
        level_limit: int = 10,
        ctx: Optional[RequestContext] = None,
    ) -> Dict:
        """Content search by pattern or keywords.

        Optimized implementation that uses agfs native grep when possible.
        The ragfs layer greps transparently over encrypted and plaintext files
        (it decrypts via account_id when an encryption layer is configured).
        Falls back to VikingFS layer implementation if native grep is unavailable.
        When engine="auto" and vikingdb is available with sufficient data,
        uses vikingdb bm25 recall + local fs precise matching.

        Args:
            uri: Viking URI
            pattern: Regular expression pattern to search for
            exclude_uri: Optional URI prefix to exclude from search
            case_insensitive: Whether to perform case-insensitive matching
            node_limit: Maximum number of results to return
            level_limit: Maximum depth level to traverse (default: 10)
            ctx: Request context
            Internal bm25 recall limit is auto-adapted from node_limit as
            min(node_limit * 5, 100000); when node_limit is unset, use 100000.

        Returns:
            Dict with matches, count, match_count, files_scanned
        """
        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.
        await self.stat(uri, ctx=ctx, skip_count=True)

        # Read engine and threshold from grep_config (ov.conf)
        engine = self.grep_config.engine if self.grep_config else "auto"
        switch_to_remote_threshold = (
            self.grep_config.switch_to_remote_threshold if self.grep_config else 10000
        )

        resolved_engine = await self._resolve_grep_engine(
            engine, uri, ctx, switch_to_remote_threshold
        )

        if resolved_engine == "fs":
            return await self._grep_fs(
                uri=uri,
                pattern=pattern,
                exclude_uri=exclude_uri,
                case_insensitive=case_insensitive,
                node_limit=node_limit,
                level_limit=level_limit,
                ctx=ctx,
            )
        else:  # "vikingdb_then_fs"
            return await self._grep_vikingdb_then_fs(
                uri=uri,
                pattern=pattern,
                exclude_uri=exclude_uri,
                case_insensitive=case_insensitive,
                node_limit=node_limit,
                level_limit=level_limit,
                ctx=ctx,
            )

    async def _resolve_grep_engine(
        self, engine: GrepEngine, uri: str, ctx, switch_to_remote_threshold: int = 10000
    ) -> str:
        """Resolve the actual grep engine to use."""
        if engine == "fs":
            return "fs"

        # auto mode: check vikingdb availability
        vector_store = self._get_vector_store()
        if not vector_store:
            return "fs"

        backend_type = getattr(vector_store, "_backend_type", "unknown")
        if backend_type not in ("volcengine", "vikingdb"):
            return "fs"

        # Check collection has content field and FullText config
        if not await self._collection_has_fulltext(vector_store, ctx):
            return "fs"

        # switch_to_remote_threshold=0 means always use vikingdb
        if switch_to_remote_threshold == 0:
            return "vikingdb_then_fs"

        # Check data volume threshold
        try:
            count = await self._get_cached_count(uri, ctx)
            if count < switch_to_remote_threshold:
                return "fs"
        except Exception:
            logger.debug(
                "grep engine=auto: count() check failed, falling back to fs", exc_info=True
            )
            return "fs"

        return "vikingdb_then_fs"

    async def _collection_has_fulltext(self, vector_store, ctx) -> bool:
        """Check if collection has content field and FullText config.

        Result is cached on the VikingFS instance since collection schema
        does not change at runtime.
        """
        if self._fulltext_available is not None:
            return self._fulltext_available
        try:
            meta = None
            if hasattr(vector_store, "get_collection_meta"):
                meta = await vector_store.get_collection_meta(ctx=ctx)
            if not meta:
                self._fulltext_available = False
                return False
            fields = meta.get("Fields", [])
            has_content = any(
                f.get("FieldName") == "content" and f.get("FieldType") == "text" for f in fields
            )
            fulltext = meta.get("FullText") or []
            has_content_fulltext = any(ft.get("Field") == "content" for ft in fulltext)
            result = has_content and has_content_fulltext
            self._fulltext_available = result
            return result
        except Exception:
            logger.debug(
                "Failed to check collection fulltext config, assuming no fulltext", exc_info=True
            )
            return False

    async def _get_cached_count(self, uri: str, ctx) -> int:
        """Get cached count of records for a URI (TTL=1h)."""
        _COUNT_CACHE_TTL = 3600
        vector_store = self._get_vector_store()

        # Include account_id in cache key for multi-tenant safety
        account_id = getattr(ctx, "account_id", None) if ctx else None
        cache_key = f"{account_id}:{uri}" if account_id else uri

        now = time.time()
        cached = self._count_cache.get(cache_key)
        if cached and (now - cached[1]) < _COUNT_CACHE_TTL:
            return cached[0]

        count = await vector_store.count(filter=PathScope("uri", uri, depth=-1), ctx=ctx)
        # Evict oldest entries if cache exceeds max size
        if len(self._count_cache) >= self._count_cache_max_size:
            oldest_keys = sorted(self._count_cache, key=lambda k: self._count_cache[k][1])
            for k in oldest_keys[: len(oldest_keys) // 2]:
                del self._count_cache[k]
        self._count_cache[cache_key] = (count, now)
        return count

    async def _grep_fs(
        self, uri, pattern, exclude_uri, case_insensitive, node_limit, level_limit, ctx
    ):
        """Filesystem grep path: prefer native agfs grep and fall back if unavailable."""
        try:
            return await self._grep_with_agfs(
                uri=uri,
                pattern=pattern,
                exclude_uri=exclude_uri,
                case_insensitive=case_insensitive,
                node_limit=node_limit,
                level_limit=level_limit,
                ctx=ctx,
            )
        except (AttributeError, AGFSNotSupportedError, NotImplementedError) as e:
            logger.debug(f"agfs grep unavailable, falling back to VikingFS implementation: {e}")

        return await self._grep_encrypted(
            uri=uri,
            pattern=pattern,
            exclude_uri=exclude_uri,
            case_insensitive=case_insensitive,
            node_limit=node_limit,
            level_limit=level_limit,
            ctx=ctx,
        )

    async def _grep_vikingdb_then_fs(
        self,
        uri,
        pattern,
        exclude_uri,
        case_insensitive,
        node_limit,
        level_limit,
        ctx,
    ):
        """VikingDB bm25 recall + local fs precise matching."""
        vector_store = self._get_vector_store()

        # Split regex alternation (e.g. "error|warning|fail") and join as a
        # single query string for bm25 search. VikingDB's standard tokenizer
        # will handle the tokenization of the query string.
        query = " ".join(kw.strip() for kw in pattern.split("|") if kw.strip())
        filter_expr = PathScope("uri", uri, depth=level_limit)
        excluded_prefix = None
        if exclude_uri:
            excluded_prefix = self._normalize_uri(exclude_uri).rstrip("/")
            self._ensure_access(excluded_prefix, ctx)
            filter_expr = And(
                [
                    filter_expr,
                    RawDSL(
                        {
                            "op": "must_not",
                            "field": "uri",
                            "conds": [excluded_prefix],
                            "para": "-d=-1",
                        }
                    ),
                ]
            )

        # Auto-adapt bm25 recall limit: recall up to 5x requested matches
        # while capping at VikingDB's max limit. If node_limit is unset,
        # use the maximum limit to avoid truncation.
        remote_return_limit = min(node_limit * 5, 100000) if node_limit else 100000

        # Step 1: vikingdb recall candidate files
        try:
            logger.debug(
                "grep vikingdb search_by_keywords request: query=%r limit=%s filter=%r "
                "output_fields=%s",
                query,
                remote_return_limit,
                filter_expr,
                ["uri"],
            )
            result = await vector_store.search_by_keywords(
                query=query,
                limit=remote_return_limit,
                filter=filter_expr,
                output_fields=["uri"],
                ctx=ctx,
            )
        except Exception as e:
            logger.warning(f"grep vikingdb step failed, falling back to fs: {e}")
            return await self._grep_fs(
                uri=uri,
                pattern=pattern,
                exclude_uri=exclude_uri,
                case_insensitive=case_insensitive,
                node_limit=node_limit,
                level_limit=level_limit,
                ctx=ctx,
            )

        candidate_uris = [r["uri"] for r in result if r.get("uri")]
        if excluded_prefix:
            candidate_uris = [
                u
                for u in candidate_uris
                if u != excluded_prefix and not u.startswith(excluded_prefix + "/")
            ]
        if not candidate_uris:
            # BM25 returned no candidates — the index confirms no matching content
            return {"matches": [], "count": 0, "match_count": 0, "files_scanned": 0}

        # Step 2: local fs precise matching on candidate files
        return await self._grep_in_files(
            candidate_uris,
            pattern,
            case_insensitive,
            node_limit,
            ctx,
        )

    async def _grep_in_files(
        self,
        file_uris: List[str],
        pattern: str,
        case_insensitive: bool,
        node_limit: Optional[int],
        ctx: Optional[RequestContext],
    ) -> Dict:
        """Execute regex matching in specified file list (vikingdb_then_fs Step 2)."""
        flags = re.IGNORECASE if case_insensitive else 0
        compiled = re.compile(pattern, flags)

        results = []
        files_scanned = 0

        for file_uri in file_uris:
            files_scanned += 1
            try:
                content_bytes = await self.read(file_uri, ctx=ctx)
                content = content_bytes.decode("utf-8", errors="replace")
            except Exception:
                continue

            for line_no, line in enumerate(content.splitlines(), 1):
                if compiled.search(line):
                    results.append({"uri": file_uri, "line": line_no, "content": line})
                    if node_limit and len(results) >= node_limit:
                        return {
                            "matches": results,
                            "count": len(results),
                            "match_count": len(results),
                            "files_scanned": files_scanned,
                        }

        return {
            "matches": results,
            "count": len(results),
            "match_count": len(results),
            "files_scanned": files_scanned,
        }

    async def _grep_with_agfs(
        self,
        uri: str,
        pattern: str,
        exclude_uri: Optional[str] = None,
        case_insensitive: bool = False,
        node_limit: Optional[int] = None,
        level_limit: int = 10,
        ctx: Optional[RequestContext] = None,
    ) -> Dict:
        """Grep using agfs native implementation.

        This is the optimized path for non-encrypted files.
        Uses agfs.grep() which performs matching on the server side.

        Prefer pushing filters down to agfs backend:
        - exclude_uri -> exclude_path
        - level_limit -> level_limit

        Args:
            uri: Viking URI
            pattern: Regular expression pattern to search for
            exclude_uri: Optional URI prefix to exclude from search
            case_insensitive: Whether to perform case-insensitive matching
            node_limit: Maximum number of results to return
            level_limit: Maximum depth level to traverse
            ctx: Request context

        Returns:
            Dict with matches, count, match_count, files_scanned
        """
        path = self._uri_to_path(uri, ctx=ctx)

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

        try:
            result = await self._async_agfs.grep(
                path=path,
                pattern=pattern,
                recursive=True,
                case_insensitive=case_insensitive,
                stream=False,
                node_limit=node_limit,
                exclude_path=excluded_path,
                level_limit=level_limit,
            )
        except (AttributeError, AGFSNotSupportedError, NotImplementedError):
            # Capability missing: let the outer caller fall back to the VikingFS implementation.
            logger.warning("agfs grep unavailable, falling back to VikingFS implementation")
            raise

        matches = result.get("matches", [])
        results = []
        files_scanned_set = set()
        real_ctx = self._ctx_or_default(ctx)

        for match in matches:
            match_file = match.get("file", "")
            if not match_file:
                continue

            agfs_file_path = self._resolve_grep_match_agfs_path(path, match_file)

            file_uri = self._path_to_uri(agfs_file_path, ctx=ctx)
            if not self._is_accessible(file_uri, real_ctx):
                continue

            files_scanned_set.add(file_uri)

            results.append(
                {
                    "line": match.get("line", match.get("line_number", 0)),
                    "uri": file_uri,
                    "content": match.get("content", ""),
                }
            )

            if node_limit and len(results) >= node_limit:
                break

        # Prefer backend-provided scanned file count if available; otherwise fall back to
        # counting files that produced at least one match (best-effort).
        backend_files_scanned = result.get("files_scanned")
        if isinstance(backend_files_scanned, int) and backend_files_scanned >= 0:
            files_scanned = (
                len(files_scanned_set) if real_ctx.actor_peer_id else backend_files_scanned
            )
        else:
            files_scanned = len(files_scanned_set)

        return {
            "matches": results,
            "count": len(results),
            "match_count": len(results),
            "files_scanned": files_scanned,
        }

    async def _grep_encrypted(
        self,
        uri: str,
        pattern: str,
        exclude_uri: Optional[str] = None,
        case_insensitive: bool = False,
        node_limit: Optional[int] = None,
        level_limit: int = 10,
        ctx: Optional[RequestContext] = None,
    ) -> Dict:
        """Grep implementation for encrypted files.

        This implementation decrypts files at VikingFS layer before matching.
        Used when encryption is enabled or when agfs.grep is not available.

        Args:
            uri: Viking URI
            pattern: Regular expression pattern to search for
            exclude_uri: Optional URI prefix to exclude from search
            case_insensitive: Whether to perform case-insensitive matching
            node_limit: Maximum number of results to return
            level_limit: Maximum depth level to traverse (default: 10)
            ctx: Request context

        Returns:
            Dict with matches, count, match_count, files_scanned
        """
        flags = re.IGNORECASE if case_insensitive else 0
        compiled_pattern = re.compile(pattern, flags)
        excluded_prefix = None
        if exclude_uri:
            excluded_prefix = self._normalize_uri(exclude_uri).rstrip("/")
            self._ensure_access(excluded_prefix, ctx)
        file_uris = await self._collect_grep_files(
            uri,
            excluded_prefix=excluded_prefix,
            level_limit=level_limit,
            ctx=ctx,
        )
        results, files_scanned = await self._grep_files_parallel(
            file_uris,
            compiled_pattern=compiled_pattern,
            node_limit=node_limit,
            ctx=ctx,
        )

        return {
            "matches": results,
            "count": len(results),
            "match_count": len(results),
            "files_scanned": files_scanned,
        }

    async def _collect_grep_files(
        self,
        uri: str,
        excluded_prefix: Optional[str],
        level_limit: int,
        ctx: Optional[RequestContext] = None,
    ) -> List[str]:
        file_uris: List[str] = []

        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)
            if excluded_prefix and (
                normalized_current_uri == excluded_prefix
                or normalized_current_uri.startswith(excluded_prefix + "/")
            ):
                logger.debug(f"Skipping excluded uri during grep: {normalized_current_uri}")
                return

            try:
                entries = await self.ls(normalized_current_uri, ctx=ctx)
            except Exception:
                return

            for entry in entries:
                entry_uri = f"{normalized_current_uri.rstrip('/')}/{entry['name']}"
                if excluded_prefix and (
                    entry_uri == excluded_prefix or entry_uri.startswith(excluded_prefix + "/")
                ):
                    logger.debug(f"Skipping excluded uri during grep: {entry_uri}")
                    continue

                if entry.get("isDir"):
                    await search_recursive(entry_uri, current_depth + 1)
                else:
                    file_uris.append(entry_uri)

        normalized_uri = self._normalize_uri(uri)
        if excluded_prefix and (
            normalized_uri == excluded_prefix or normalized_uri.startswith(excluded_prefix + "/")
        ):
            logger.debug(f"Skipping excluded uri during grep: {normalized_uri}")
            return file_uris
        try:
            root_stat = await self.stat(normalized_uri, ctx=ctx)
        except Exception:
            return file_uris
        if not root_stat.get("isDir", False):
            file_uris.append(normalized_uri)
            return file_uris

        await search_recursive(uri, 0)
        return file_uris

    async def _grep_files_parallel(
        self,
        file_uris: List[str],
        compiled_pattern: re.Pattern,
        node_limit: Optional[int],
        ctx: Optional[RequestContext] = None,
    ) -> tuple[List[Dict[str, Any]], int]:
        results: List[Dict[str, Any]] = []
        files_scanned = 0
        for start in range(0, len(file_uris), _DEFAULT_GREP_FILE_CONCURRENCY):
            batch_uris = file_uris[start : start + _DEFAULT_GREP_FILE_CONCURRENCY]
            batch_jobs = [
                self._grep_single_file(entry_uri, compiled_pattern, ctx) for entry_uri in batch_uris
            ]
            batch_results = await asyncio.gather(*batch_jobs)
            for matches, scanned_count in batch_results:
                files_scanned += scanned_count
                for match in matches:
                    results.append(match)
                    if node_limit and len(results) >= node_limit:
                        return results, files_scanned

        return results, files_scanned

    async def _grep_single_file(
        self,
        entry_uri: str,
        compiled_pattern: re.Pattern,
        ctx: Optional[RequestContext] = None,
    ) -> tuple[List[Dict[str, Any]], int]:
        try:
            content = await self.read(entry_uri, ctx=ctx)
            if isinstance(content, bytes):
                content = content.decode("utf-8", errors="replace")

            matches: List[Dict[str, Any]] = []
            lines = content.split("\n")
            for line_num, line in enumerate(lines, 1):
                if compiled_pattern.search(line):
                    matches.append(
                        {
                            "line": line_num,
                            "uri": entry_uri,
                            "content": line,
                        }
                    )
            return matches, 1
        except Exception as e:
            logger.debug(f"Failed to grep {entry_uri}: {e}")
            return [], 1

    def _resolve_grep_match_agfs_path(self, base_path: str, match_file: str) -> str:
        """Resolve a grep match path (relative to query root) into a full AGFS path."""
        if match_file == ".":
            return base_path
        return f"{base_path.rstrip('/')}/{match_file.lstrip('/')}"

    def _calculate_grep_match_depth(self, match_file: str) -> int:
        """Calculate relative depth from a grep result path relative to the query root."""
        if not match_file or match_file == ".":
            return 0
        return len([part for part in match_file.split("/") if part])

    async def stat(
        self, uri: str, ctx: Optional[RequestContext] = None, skip_count: bool = False
    ) -> Dict[str, Any]:
        """
        File/directory information.

        example: {'name': 'resources', 'size': 128, 'mode': 2147484141, 'modTime': '2026-02-10T21:26:02.934376379+08:00', 'isDir': True, 'isLocked': False, 'count': 42, 'meta': {'Name': 'localfs', 'Type': 'local', 'Content': {'local_path': '...'}}}

        Extra field:
            isLocked (bool): Whether the path is currently held by a path lock
                (either the path itself or any ancestor directory). Returns
                False when the LockManager is not initialized or the lookup
                fails.
            count (int): For directories, the number of nodes in the vector index
                under this directory (including subdirectories). For files, this
                field is not included.

        Args:
            uri: Viking URI
            ctx: Request context
            skip_count: If True, skip the vector_store.count() call for directories.
                Use this when the count field is not needed (e.g. in grep) to avoid
                an extra VikingDB API call.
        """
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        path = primary_path
        last_not_found: Optional[Exception] = None
        for candidate_path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, candidate_path, primary_path, real_ctx):
                continue
            try:
                result = await self._async_agfs.stat(candidate_path)
                path = candidate_path
                break
            except Exception as exc:
                if is_not_found_error(exc):
                    last_not_found = exc
                    continue
                raise
        else:
            if self._is_legacy_session_root_uri(uri):
                now = datetime.now(timezone.utc).isoformat()
                return {
                    "name": "session",
                    "size": 0,
                    "mode": 0o755,
                    "modTime": now,
                    "isDir": True,
                    "isLocked": False,
                }
            raise NotFoundError(uri, "file") from last_not_found
        if isinstance(result, dict):
            result["isLocked"] = await self._is_path_locked_async(path)
            # Add count for directories if vector store available
            if not skip_count and result.get("isDir", False):
                try:
                    vector_store = self._get_vector_store()
                    if vector_store:
                        target_canonical_uri = canonicalize_uri(
                            self._path_to_uri(path, ctx=real_ctx), real_ctx
                        )
                        if not may_include_hidden_actor_peers(target_canonical_uri, real_ctx):
                            filter_expr = PathScope("uri", target_canonical_uri, depth=-1)
                            result["count"] = await vector_store.count(
                                filter=filter_expr,
                                ctx=real_ctx,
                            )
                except Exception as e:
                    logger.warning(f"[VikingFS] Failed to count nodes for directory stat: {e}")
        return result

    async def _is_path_locked_async(self, path: str) -> bool:
        """Best-effort async path-lock lookup; returns False when LockManager is absent."""
        try:
            from openviking.storage.transaction import get_lock_manager

            return await get_lock_manager().is_path_locked_async(path)
        except Exception:
            return False

    async def exists(self, uri: str, ctx: Optional[RequestContext] = None) -> bool:
        """Check if a URI exists.

        Args:
            uri: Viking URI
            ctx: Request context

        Returns:
            bool: True if the URI exists, False otherwise
        """
        try:
            await self.stat(uri, ctx=ctx)
            return True
        except Exception:
            return False

    async def glob(
        self,
        pattern: str,
        uri: str = "viking://",
        node_limit: Optional[int] = None,
        ctx: Optional[RequestContext] = None,
    ) -> Dict:
        """File pattern matching, supports **/*.md recursive."""
        entries = await self.tree(uri, node_limit=1000000, level_limit=None, ctx=ctx)
        matches = []
        for entry in entries:
            rel_path = entry.get("rel_path", "")
            if PurePath(rel_path).match(pattern):
                matches.append(entry["uri"])
        # Now apply node limit to the filtered matches
        if node_limit is not None and node_limit > 0:
            matches = matches[:node_limit]
        return {"matches": matches, "count": len(matches)}

    async def _batch_fetch_abstracts(
        self,
        entries: List[Dict[str, Any]],
        abs_limit: int,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Batch fetch abstracts for entries using a fixed-size worker pool.

        Non-directory entries receive an empty abstract immediately.
        Directory entries are processed concurrently via a worker pool,
        using _read_abstract_for_known_dir to skip redundant stat() calls.

        Args:
            entries: List of entries to fetch abstracts for
            abs_limit: Maximum length for abstract truncation
        """
        dir_jobs = []
        for index, entry in enumerate(entries):
            if not entry.get("isDir", False):
                entry["abstract"] = ""
                continue
            dir_jobs.append((index, entry))

        if not dir_jobs:
            return

        worker_count = min(_ABSTRACT_WORKER_COUNT, len(dir_jobs))

        cursor = 0
        cursor_lock = asyncio.Lock()
        results: Dict[int, str] = {}

        async def worker() -> None:
            nonlocal cursor
            while True:
                async with cursor_lock:
                    if cursor >= len(dir_jobs):
                        return
                    index, entry = dir_jobs[cursor]
                    cursor += 1

                try:
                    abstract = await self._read_abstract_for_known_dir(entry["uri"], ctx=ctx)
                except Exception:
                    abstract = "[.abstract.md is not ready]"

                results[index] = abstract

        await asyncio.gather(*(worker() for _ in range(worker_count)))

        for index, abstract in results.items():
            if len(abstract) > abs_limit:
                abstract = abstract[: abs_limit - 3] + "..."
            entries[index]["abstract"] = abstract

    async def tree(
        self,
        uri: str = "viking://",
        output: str = "original",
        abs_limit: int = 256,
        show_all_hidden: bool = False,
        node_limit: Optional[int] = 1000,
        level_limit: Optional[int] = 3,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """
        Recursively list all contents (includes rel_path).

        Args:
            uri: Viking URI
            output: str = "original" or "agent"
            abs_limit: int = 256 (for agent output abstract truncation)
            show_all_hidden: bool = False (list all hidden files, like -a)
            node_limit: int | None = 1000 (maximum number of nodes to list, None means unlimited)
            level_limit: int | None = 3 (maximum depth level to traverse, None means unlimited)

        output="original"
        [{'name': '.abstract.md', 'size': 100, 'mode': 420, 'modTime': '2026-02-11T16:52:16.256334192+08:00', 'isDir': False, 'rel_path': '.abstract.md', 'uri': 'viking://resources...'}]

        output="agent"
        [{'uri': 'viking://resources...', 'size': 100, 'isDir': False, 'modTime': '2026-02-11T08:52:16.256Z', 'rel_path': '.abstract.md', 'abstract': "..."}]
        """
        self._ensure_access(uri, ctx)
        if output == "original":
            return await self._tree_original(uri, show_all_hidden, node_limit, level_limit, ctx=ctx)
        elif output == "agent":
            return await self._tree_agent(
                uri, abs_limit, show_all_hidden, node_limit, level_limit, ctx=ctx
            )
        else:
            raise ValueError(f"Invalid output format: {output}")

    async def _tree_original(
        self,
        uri: str,
        show_all_hidden: bool = False,
        node_limit: Optional[int] = 1000,
        level_limit: Optional[int] = 3,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """Recursively list all contents (original format)."""
        result = []
        async for entry, entry_uri in self._iter_visible_tree_entries(
            uri,
            show_all_hidden=show_all_hidden,
            node_limit=node_limit,
            level_limit=level_limit,
            ctx=ctx,
        ):
            info = entry["info"]
            new_entry = dict(entry.get("extra", {}))
            new_entry.update(
                {
                    "name": info["name"],
                    "size": info["size"],
                    "mode": info["mode"],
                    "modTime": info["modTime"],
                    "isDir": info["isDir"],
                    "rel_path": entry["rel_path"],
                    "uri": entry_uri,
                }
            )
            result.append(new_entry)
        return result

    async def _tree_agent(
        self,
        uri: str,
        abs_limit: int,
        show_all_hidden: bool = False,
        node_limit: Optional[int] = 1000,
        level_limit: Optional[int] = 3,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """Recursively list all contents (agent format with abstracts)."""
        result = []

        async for entry, entry_uri in self._iter_visible_tree_entries(
            uri,
            show_all_hidden=show_all_hidden,
            node_limit=node_limit,
            level_limit=level_limit,
            ctx=ctx,
        ):
            info = entry["info"]
            is_dir = info["isDir"]
            result.append(
                {
                    "uri": entry_uri,
                    "size": 0 if is_dir else info["size"],
                    "isDir": is_dir,
                    "modTime": format_iso8601(parse_iso_datetime(info["modTime"])),
                    "rel_path": entry["rel_path"],
                }
            )

        await self._batch_fetch_abstracts(result, abs_limit, ctx=ctx)

        return result

    # ========== VikingFS Specific Capabilities ==========

    async def _read_abstract_file(
        self,
        path: str,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> str:
        """Read and decrypt/decode .abstract.md from a known directory path.

        Does NOT perform stat or isDir check -- caller is responsible for
        ensuring the path points to a directory.
        """
        file_path = f"{path}/.abstract.md"
        try:
            content_bytes = self._handle_agfs_read(await self._async_agfs.read(file_path))
        except Exception as exc:
            if not is_not_found_error(exc):
                mapped = map_exception(exc, resource=uri)
                if mapped is not None:
                    raise mapped from exc
                raise
            return f"# {uri} [Directory abstract is not ready]"

        return self._decode_bytes(content_bytes)

    async def _read_abstract_for_known_dir(
        self,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> str:
        """Read .abstract.md for a directory that is already known to be a directory.

        Bypasses stat() and isDir check. Caller (i.e. _batch_fetch_abstracts)
        must guarantee that the URI points to a directory.
        """
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        for path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, path, primary_path, real_ctx):
                continue
            try:
                if not await self._agfs_path_exists(path):
                    continue
                return await self._read_abstract_file(path, uri, ctx=ctx)
            except Exception as exc:
                if is_not_found_error(exc):
                    continue
                raise
        return f"# {uri} [Directory abstract is not ready]"

    async def abstract(
        self,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> str:
        """Read directory's L0 summary (.abstract.md).

        If the caller points to a file, its parent directory is used instead so
        the endpoint remains usable for both file and directory URIs.
        """
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        path = primary_path
        last_exc: Optional[Exception] = None
        for candidate_path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, candidate_path, primary_path, real_ctx):
                continue
            try:
                info = await self._async_agfs.stat(candidate_path)
                path = candidate_path
                break
            except Exception as exc:
                if is_not_found_error(exc):
                    last_exc = exc
                    continue
                mapped = map_exception(exc, resource=uri)
                if mapped is not None:
                    raise mapped from exc
                raise
        else:
            if last_exc is not None:
                mapped = map_exception(last_exc, resource=uri)
                if mapped is not None:
                    raise mapped from last_exc
            raise NotFoundError(uri, "directory") from last_exc
        if not info.get("isDir", info.get("is_dir")):
            parent_path = path.rsplit("/", 1)[0] or "/"
            parent_uri = self._path_to_uri(parent_path, ctx=ctx)
            logger.info(
                "content/abstract: %s is a file, falling back to parent directory %s",
                uri,
                parent_uri,
            )
            return await self.abstract(parent_uri, ctx=ctx)
        return await self._read_abstract_file(path, uri, ctx=ctx)

    async def overview(
        self,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> str:
        """Read directory's L1 overview (.overview.md).

        If the caller points to a file, its parent directory is used instead so
        the endpoint remains usable for both file and directory URIs.
        """
        self._ensure_access(uri, ctx=ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        path = primary_path
        last_exc: Optional[Exception] = None
        for candidate_path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, candidate_path, primary_path, real_ctx):
                continue
            try:
                info = await self._async_agfs.stat(candidate_path)
                path = candidate_path
                break
            except Exception as exc:
                if is_not_found_error(exc):
                    last_exc = exc
                    continue
                mapped = map_exception(exc, resource=uri)
                if mapped is not None:
                    raise mapped from exc
                raise
        else:
            if last_exc is not None:
                mapped = map_exception(last_exc, resource=uri)
                if mapped is not None:
                    raise mapped from last_exc
            raise NotFoundError(uri, "directory") from last_exc
        if not info.get("isDir", info.get("is_dir")):
            parent_path = path.rsplit("/", 1)[0] or "/"
            parent_uri = self._path_to_uri(parent_path, ctx=ctx)
            logger.info(
                "content/overview: %s is a file, falling back to parent directory %s",
                uri,
                parent_uri,
            )
            return await self.overview(parent_uri, ctx=ctx)
        file_path = f"{path}/.overview.md"
        try:
            content_bytes = self._handle_agfs_read(await self._async_agfs.read(file_path))
        except Exception as exc:
            if not is_not_found_error(exc):
                mapped = map_exception(exc, resource=uri)
                if mapped is not None:
                    raise mapped from exc
                raise
            # Fallback to default if .overview.md doesn't exist
            return f"# {uri}\n\n[Directory overview is not ready]"

        return self._decode_bytes(content_bytes)

    async def relations(
        self,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """Get relation list.

        Returns: [{"uri": "...", "reason": "..."}, ...]
        """
        self._ensure_access(uri, ctx)
        entries = await self.get_relation_table(uri, ctx=ctx)
        result = []
        for entry in entries:
            for u in entry.uris:
                if self._is_accessible(u, self._ctx_or_default(ctx)):
                    result.append({"uri": u, "reason": entry.reason})
        return result

    async def find(
        self,
        query: str,
        target_uri: Union[str, List[str]] = "",
        limit: int = 10,
        score_threshold: Optional[float] = None,
        filter: Optional[Dict] = None,
        ctx: Optional[RequestContext] = None,
        level: Optional[List[int]] = None,
    ):
        """Semantic search.

        Args:
            query: Search query
            target_uri: Target directory URI(s), supports str or List[str]
            limit: Return count
            score_threshold: Score threshold
            filter: Metadata filter

        Returns:
            FindResult
        """
        _ensure_non_empty_search_query(query)
        telemetry = get_current_telemetry()
        from openviking.retrieve.hierarchical_retriever import HierarchicalRetriever
        from openviking_cli.retrieve import (
            ContextType,
            FindResult,
            TypedQuery,
        )

        real_ctx = self._ctx_or_default(ctx)
        retrieval_targets = resolve_retrieval_targets(target_uri, real_ctx)

        for target_dir in retrieval_targets.target_directories:
            self._ensure_access(target_dir, ctx)

        storage = self._get_vector_store()
        if not storage:
            raise RuntimeError("Vector store not initialized. Call OpenViking.initialize() first.")

        embedder = self._get_embedder()
        if not embedder:
            raise RuntimeError("Embedder not configured.")

        retriever = HierarchicalRetriever(
            storage=storage,
            embedder=embedder,
            rerank_config=self.rerank_config,
            retrieval_config=self.retrieval_config,
        )

        typed_query = TypedQuery(
            query=query,
            context_type=None,
            intent="",
            target_directories=retrieval_targets.target_directories,
        )

        logger.debug(
            "[VikingFS.find] Calling retriever.retrieve with "
            f"ctx.account_id={real_ctx.account_id}, ctx.user={real_ctx.user}"
        )

        result = await retriever.retrieve(
            typed_query,
            ctx=real_ctx,
            limit=limit,
            score_threshold=score_threshold,
            scope_dsl=filter,
            level=level,
        )

        # Convert QueryResult to FindResult
        memories, resources, skills = [], [], []
        for ctx in result.matched_contexts:
            if ctx.context_type == ContextType.MEMORY:
                memories.append(ctx)
            elif ctx.context_type == ContextType.RESOURCE:
                resources.append(ctx)
            elif ctx.context_type == ContextType.SKILL:
                skills.append(ctx)

        find_result = FindResult(
            memories=memories,
            resources=resources,
            skills=skills,
        )
        telemetry.set("vector.returned", find_result.total)
        return find_result

    async def search(
        self,
        query: str,
        target_uri: Union[str, List[str]] = "",
        session_info: Optional[Dict] = None,
        limit: int = 10,
        score_threshold: Optional[float] = None,
        filter: Optional[Dict] = None,
        ctx: Optional[RequestContext] = None,
        level: Optional[List[int]] = None,
    ):
        """Complex search with session context.

        Args:
            query: Search query
            target_uri: Target directory URI(s), supports str or List[str]
            session_info: Session information
            limit: Return count
            filter: Metadata filter

        Returns:
            FindResult
        """
        _ensure_non_empty_search_query(query)
        telemetry = get_current_telemetry()
        from openviking.retrieve.hierarchical_retriever import HierarchicalRetriever
        from openviking.retrieve.intent_analyzer import IntentAnalyzer
        from openviking_cli.retrieve import (
            ContextType,
            FindResult,
            QueryPlan,
            TypedQuery,
        )

        real_ctx = self._ctx_or_default(ctx)
        retrieval_targets = resolve_retrieval_targets(target_uri, real_ctx)
        primary_target_uri = retrieval_targets.first_explicit_directory

        session_summary = (
            str(session_info.get("latest_archive_overview") or "") if session_info else ""
        )
        current_messages = session_info.get("current_messages") if session_info else None

        query_plan: Optional[QueryPlan] = None
        for target_dir in retrieval_targets.target_directories:
            self._ensure_access(target_dir, ctx)

        # When target_uri exists, read its abstract as optional query-planning context.
        target_abstract = ""
        if primary_target_uri:
            try:
                with telemetry.measure("search.target_abstract"):
                    target_abstract = await self.abstract(primary_target_uri, ctx=ctx)
            except Exception:
                target_abstract = ""

        # With session context: intent analysis
        if session_summary or current_messages:
            analyzer = IntentAnalyzer(max_recent_messages=5)
            with telemetry.measure("search.intent_analysis"):
                query_plan = await analyzer.analyze(
                    compression_summary=session_summary or "",
                    messages=current_messages or [],
                    current_message=query,
                    target_abstract=target_abstract,
                )
            typed_queries = query_plan.queries
            for tq in typed_queries:
                tq.target_directories = retrieval_targets.target_directories
        else:
            # No session context: create query directly
            typed_queries = [
                TypedQuery(
                    query=query,
                    context_type=None,
                    intent="",
                    priority=1,
                    target_directories=retrieval_targets.target_directories,
                )
            ]
        telemetry.set("search.typed_queries_count", len(typed_queries))

        # Concurrent execution
        storage = self._get_vector_store()
        embedder = self._get_embedder()
        retriever = HierarchicalRetriever(
            storage=storage,
            embedder=embedder,
            rerank_config=self.rerank_config,
            retrieval_config=self.retrieval_config,
        )

        async def _execute(tq: TypedQuery):
            real_ctx = self._ctx_or_default(ctx)
            logger.debug(
                "[VikingFS.search._execute] Calling retriever.retrieve with "
                f"ctx.account_id={real_ctx.account_id}, ctx.user={real_ctx.user}"
            )
            return await retriever.retrieve(
                tq,
                ctx=real_ctx,
                limit=limit,
                score_threshold=score_threshold,
                scope_dsl=filter,
                level=level,
            )

        query_results = await asyncio.gather(*[_execute(tq) for tq in typed_queries])

        # Aggregate results to FindResult
        memories, resources, skills = [], [], []
        for result in query_results:
            for ctx in result.matched_contexts:
                if ctx.context_type == ContextType.MEMORY:
                    memories.append(ctx)
                elif ctx.context_type == ContextType.RESOURCE:
                    resources.append(ctx)
                elif ctx.context_type == ContextType.SKILL:
                    skills.append(ctx)

        find_result = FindResult(
            memories=memories,
            resources=resources,
            skills=skills,
            query_plan=query_plan,
            query_results=query_results,
        )
        telemetry.set("vector.returned", find_result.total)
        return find_result

    # ========== Relation Management ==========

    async def link(
        self,
        from_uri: str,
        uris: Union[str, List[str]],
        reason: str = "",
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Create relation (maintained in .relations.json)."""
        if isinstance(uris, str):
            uris = [uris]
        self._ensure_mutable_access(from_uri, ctx)
        for uri in uris:
            self._ensure_access(uri, ctx)

        from_path = self._uri_to_path(from_uri, ctx=ctx)

        entries = await self._read_relation_table(from_path, ctx=ctx)
        existing_ids = {e.id for e in entries}

        link_id = next(f"link_{i}" for i in range(1, 10000) if f"link_{i}" not in existing_ids)

        entries.append(RelationEntry(id=link_id, uris=uris, reason=reason))

        await self._write_relation_table(from_path, entries, ctx=ctx)
        logger.debug(f"[VikingFS] Created link: {from_uri} -> {uris}")

    async def unlink(
        self,
        from_uri: str,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Delete relation."""
        self._ensure_mutable_access(from_uri, ctx)
        self._ensure_access(uri, ctx)
        from_path = self._uri_to_path(from_uri, ctx=ctx)

        try:
            entries = await self._read_relation_table(from_path, ctx=ctx)

            entry_to_modify = None
            for entry in entries:
                if uri in entry.uris:
                    entry_to_modify = entry
                    break

            if not entry_to_modify:
                logger.debug(f"[VikingFS] URI not found in relations: {uri}")
                return

            entry_to_modify.uris.remove(uri)

            if not entry_to_modify.uris:
                entries.remove(entry_to_modify)
                logger.debug(f"[VikingFS] Removed empty entry: {entry_to_modify.id}")

            await self._write_relation_table(from_path, entries, ctx=ctx)
            logger.debug(f"[VikingFS] Removed link: {from_uri} -> {uri}")

        except Exception as e:
            logger.error(f"[VikingFS] Failed to unlink {from_uri} -> {uri}: {e}")
            raise IOError(f"Failed to unlink: {e}")

    async def get_relation_table(
        self, uri: str, ctx: Optional[RequestContext] = None
    ) -> List[RelationEntry]:
        """Get relation table."""
        self._ensure_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        return await self._read_relation_table(path, ctx=ctx)

    # ========== Tree Traversal (Refactored) ==========

    def _is_name_visible_at_path(self, name: str, parent_path: str) -> bool:
        """Check if name would appear in _ls_entries(parent_path).

        At account root (/local/{account}), uses LISTABLE_SCOPES whitelist.
        At other levels, uses the shared storage internal-name blacklist.
        """
        parts = [p for p in parent_path.strip("/").split("/") if p]
        if len(parts) == 2 and parts[0] == "local":
            return name in VikingURI.LISTABLE_SCOPES
        return name not in STORAGE_INTERNAL_ENTRY_NAMES

    def _ancestor_is_filtered(self, entry_path: str, base_path: str) -> bool:
        """Check if any ancestor directory of entry_path would be filtered by _ls_entries.

        Walks from base_path (exclusive) to entry's parent directory (exclusive),
        checking each component against _is_name_visible_at_path.
        """
        base_parts = [p for p in base_path.strip("/").split("/") if p]
        entry_parts = [p for p in entry_path.strip("/").split("/") if p]

        for i in range(len(base_parts), len(entry_parts) - 1):
            name = entry_parts[i]
            parent_parts = entry_parts[:i]
            parent_path = "/" + "/".join(parent_parts) if parent_parts else "/"
            if not self._is_name_visible_at_path(name, parent_path):
                return True
        return False

    def _is_tree_entry_visible(
        self, entry: Dict[str, Any], base_path: str, ctx: RequestContext
    ) -> bool:
        """Check visibility for a single TreeEntry returned by Rust tree_directory.

        Applies three layers of filtering:
        1. Ancestor chain — if any ancestor directory would be filtered by _ls_entries,
           all descendants are invisible.
        2. Self — the entry's own name must pass _ls_entries at its parent level.
        3. ACL — the entry must be accessible by the requesting context.
        """
        entry_path = entry["path"]

        if self._ancestor_is_filtered(entry_path, base_path):
            return False

        entry_parts = [p for p in entry_path.strip("/").split("/") if p]
        if entry_parts:
            name = entry_parts[-1]
            parent_parts = entry_parts[:-1]
            parent_path = "/" + "/".join(parent_parts) if parent_parts else "/"
            if not self._is_name_visible_at_path(name, parent_path):
                return False

        uri = self._path_to_uri(entry_path, ctx=ctx)
        if not self._is_accessible(uri, ctx):
            return False

        return True

    # Over-fetch multiplier for bounded tree traversal. When a node_limit is
    # set, we push down node_limit * this factor as the raw-node limit to Rust,
    # leaving headroom for ACL/internal-name filtering before re-fetching.
    _TREE_OVERFETCH_FACTOR = 4

    async def _iter_visible_tree_entries(
        self,
        uri: str,
        show_all_hidden: bool = False,
        node_limit: Optional[int] = None,
        level_limit: Optional[int] = None,
        ctx: Optional[RequestContext] = None,
    ):
        """Shared generator: fetch raw TreeEntry list from Rust, yield (entry, uri) tuples.

        node_limit counts ACL-visible entries (see design §6.5), so the user's
        node_limit cannot be pushed directly to Rust — doing so would truncate
        before filtering and drop entries that should be visible.

        To keep memory bounded without changing that semantic, we push down an
        *amplified* raw-node limit (node_limit * _TREE_OVERFETCH_FACTOR). If ACL
        filtering leaves fewer than node_limit visible entries while Rust still
        returned a full page (i.e. more raw nodes may exist), we double the raw
        limit and re-fetch. Because Rust truncates a deterministic sorted prefix,
        this yields exactly the same result as an unbounded fetch, while avoiding
        materializing the entire prefix in the common case.

        When node_limit is None (full-tree callers), no limit is pushed down.
        level_limit IS always passed to Rust.
        """
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        path: Optional[str] = None
        for candidate_path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, candidate_path, primary_path, real_ctx):
                continue
            if await self._agfs_path_exists(candidate_path):
                path = candidate_path
                break
        if path is None:
            if self._is_legacy_session_root_uri(uri):
                return
            raise NotFoundError(uri, "directory")

        if node_limit is None:
            raw_limit: Optional[int] = None
        else:
            raw_limit = max(node_limit * self._TREE_OVERFETCH_FACTOR, node_limit)

        while True:
            raw_entries = await self._async_agfs.tree_directory(
                path,
                show_hidden=show_all_hidden,
                node_limit=raw_limit,
                level_limit=level_limit,
            )

            visible: List[tuple] = []
            for entry in raw_entries:
                if node_limit is not None and len(visible) >= node_limit:
                    break
                if not self._is_tree_entry_visible(entry, path, real_ctx):
                    continue
                if not await self._read_path_visible(uri, entry["path"], primary_path, real_ctx):
                    continue
                entry_uri = self._alias_uri_for_path(
                    request_uri=uri,
                    base_path=path,
                    entry_path=entry["path"],
                    ctx=ctx,
                )
                visible.append((entry, entry_uri))

            # If we still lack enough visible entries but Rust returned a full
            # page (raw_limit reached), more raw nodes may exist — re-fetch with
            # a doubled limit. Otherwise Rust is exhausted and we yield as-is.
            need_more = (
                node_limit is not None
                and len(visible) < node_limit
                and raw_limit is not None
                and len(raw_entries) >= raw_limit
            )
            if need_more:
                raw_limit *= 2
                continue

            for item in visible:
                yield item
            return

    # ========== URI Conversion ==========

    # Maximum bytes for a single filename component (filesystem limit is typically 255)
    _MAX_FILENAME_BYTES = 255

    @staticmethod
    def _shorten_component(component: str, max_bytes: int = 255) -> str:
        """Shorten a path component if its UTF-8 encoding exceeds max_bytes."""
        if len(component.encode("utf-8")) <= max_bytes:
            return component
        hash_suffix = hashlib.sha256(component.encode("utf-8")).hexdigest()[:8]
        # Trim to fit within max_bytes after adding hash suffix
        prefix = component
        target = max_bytes - len(f"_{hash_suffix}".encode("utf-8"))
        while len(prefix.encode("utf-8")) > target and prefix:
            prefix = prefix[:-1]
        return f"{prefix}_{hash_suffix}"

    def _uri_to_path(self, uri: str, ctx: Optional[RequestContext] = None) -> str:
        """Map virtual URI to account-isolated AGFS path.

        Pure prefix replacement: viking://{remainder} -> /local/{account_id}/{remainder}.
        No implicit space injection — URIs must include space segments explicitly.
        """
        real_ctx = self._ctx_or_default(ctx)
        account_id = real_ctx.account_id
        normalized_uri, legacy_parts = self._normalized_uri_parts(uri)
        if legacy_parts and legacy_parts[0] == "agent" and self._is_legacy_agent_id_uri(uri):
            # Old format: viking://agent/{agent_id}/... — direct mapping for read-only compat
            safe_parts = [
                self._shorten_component(p, self._MAX_FILENAME_BYTES) for p in legacy_parts
            ]
            return f"/local/{account_id}/{'/'.join(safe_parts)}"
        canonical_uri = canonicalize_uri(uri, real_ctx)
        _, parts = self._normalized_uri_parts(canonical_uri)
        if not parts:
            return f"/local/{account_id}"

        safe_parts = [self._shorten_component(p, self._MAX_FILENAME_BYTES) for p in parts]
        return f"/local/{account_id}/{'/'.join(safe_parts)}"

    def _legacy_session_path(self, uri: str, ctx: Optional[RequestContext] = None) -> str:
        """Map a legacy viking://session URI to its pre-user-namespace path."""
        real_ctx = self._ctx_or_default(ctx)
        _, parts = self._normalized_uri_parts(uri)
        safe_parts = [self._shorten_component(p, self._MAX_FILENAME_BYTES) for p in parts]
        return f"/local/{real_ctx.account_id}/{'/'.join(safe_parts)}"

    def _legacy_current_user_session_path(
        self, uri: str, ctx: Optional[RequestContext] = None
    ) -> Optional[str]:
        """Return the legacy nested /session/{user_id}/{session_id} candidate."""
        real_ctx = self._ctx_or_default(ctx)
        _, parts = self._normalized_uri_parts(uri)
        if len(parts) <= 1 or parts[0] != "session":
            return None
        nested_parts = ["session", real_ctx.user.user_id, *parts[1:]]
        safe_parts = [self._shorten_component(p, self._MAX_FILENAME_BYTES) for p in nested_parts]
        return f"/local/{real_ctx.account_id}/{'/'.join(safe_parts)}"

    def _is_legacy_session_uri(self, uri: str) -> bool:
        _, parts = self._normalized_uri_parts(uri)
        return bool(parts and parts[0] == "session")

    def _is_legacy_session_root_uri(self, uri: str) -> bool:
        _, parts = self._normalized_uri_parts(uri)
        return parts == ["session"]

    def _is_legacy_agent_id_uri(self, uri: str) -> bool:
        _, parts = self._normalized_uri_parts(uri)
        return bool(
            parts
            and parts[0] == "agent"
            and len(parts) >= 2
            and parts[1] not in {"skills", "endpoints", "tools", "payments"}
        )

    def _read_paths(self, uri: str, ctx: Optional[RequestContext] = None) -> List[str]:
        """Return read candidates for a URI, including legacy alias fallbacks."""
        paths = [self._uri_to_path(uri, ctx=ctx)]

        if self._is_legacy_session_uri(uri):
            for candidate in (
                self._legacy_session_path(uri, ctx=ctx),
                self._legacy_current_user_session_path(uri, ctx=ctx),
            ):
                if candidate and candidate not in paths:
                    paths.append(candidate)
        return paths

    async def _read_path_visible(
        self,
        request_uri: str,
        path: str,
        primary_path: str,
        ctx: RequestContext,
    ) -> bool:
        if path == primary_path:
            return True
        if self._is_legacy_session_uri(request_uri):
            return await self._legacy_session_path_visible(path, ctx)
        return True

    def _alias_uri_for_path(
        self,
        *,
        request_uri: str,
        base_path: str,
        entry_path: str,
        ctx: Optional[RequestContext],
    ) -> str:
        normalized_request, request_parts = self._normalized_uri_parts(request_uri)
        # Only legacy namespaces need alias remapping:
        # - Old format viking://agent/{agent_id}/...
        # - viking://session/...
        is_legacy_namespace = request_parts and (
            request_parts[0] == "session"
            or (request_parts[0] == "agent" and self._is_legacy_agent_id_uri(request_uri))
        )
        if not is_legacy_namespace:
            return self._path_to_uri(entry_path, ctx=ctx)
        base = base_path.rstrip("/")
        rel_path = entry_path[len(base) :].strip("/") if entry_path.startswith(base) else ""
        if not rel_path:
            return normalized_request.rstrip("/")
        return f"{normalized_request.rstrip('/')}/{rel_path}"

    async def _agfs_path_exists(self, path: str) -> bool:
        try:
            await self._async_agfs.stat(path)
            return True
        except Exception as exc:
            if is_not_found_error(exc):
                return False
            raise

    async def _looks_like_legacy_session_dir(self, path: str) -> bool:
        for leaf in (".meta.json", "messages.jsonl", "history", "tool-results", "tools"):
            if await self._agfs_path_exists(f"{path}/{leaf}"):
                return True
        return False

    async def _legacy_session_owner(self, session_root_path: str) -> str:
        try:
            raw = self._handle_agfs_read(
                await self._async_agfs.read(f"{session_root_path}/.meta.json")
            )
        except Exception as exc:
            if is_not_found_error(exc):
                return ""
            raise
        try:
            data = json.loads(self._decode_bytes(raw))
        except json.JSONDecodeError:
            return ""
        if not isinstance(data, dict):
            return ""
        for key in ("created_by_user_id", "user_id", "owner_user_id", "created_by"):
            value = data.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
        return ""

    async def _legacy_session_visible(
        self,
        session_root_path: str,
        ctx: RequestContext,
        *,
        owner_hint: Optional[str] = None,
    ) -> bool:
        if ctx.role == Role.ROOT:
            return True
        owner = await self._legacy_session_owner(session_root_path)
        if owner:
            return owner == ctx.user.user_id
        if owner_hint:
            return owner_hint == ctx.user.user_id
        return True

    async def _legacy_session_path_visible(self, path: str, ctx: RequestContext) -> bool:
        parts = [p for p in path.strip("/").split("/") if p]
        try:
            session_index = parts.index("session")
        except ValueError:
            return True
        suffix = parts[session_index + 1 :]
        if not suffix:
            return True

        root_prefix = "/" + "/".join(parts[: session_index + 1])
        direct_root = f"{root_prefix}/{suffix[0]}"
        if await self._looks_like_legacy_session_dir(direct_root):
            return await self._legacy_session_visible(direct_root, ctx)

        if len(suffix) >= 2:
            nested_root = f"{root_prefix}/{suffix[0]}/{suffix[1]}"
            if await self._looks_like_legacy_session_dir(nested_root):
                return await self._legacy_session_visible(
                    nested_root,
                    ctx,
                    owner_hint=suffix[0],
                )
        return True

    async def _legacy_session_root_items(
        self,
        path: str,
        ctx: RequestContext,
    ) -> List[tuple[Dict[str, Any], str]]:
        try:
            entries = await self._ls_entries(path)
        except Exception as exc:
            if is_not_found_error(exc):
                return []
            raise

        items: List[tuple[Dict[str, Any], str]] = []
        for entry in entries:
            name = entry.get("name", "")
            if not name or name in {".", ".."} or not entry.get("isDir"):
                continue
            child_path = f"{path.rstrip('/')}/{name}"
            if await self._looks_like_legacy_session_dir(child_path):
                if await self._legacy_session_visible(child_path, ctx):
                    items.append((entry, f"viking://session/{name}"))
                continue

            if ctx.role != Role.ROOT and name != ctx.user.user_id:
                continue
            try:
                nested_entries = await self._ls_entries(child_path)
            except Exception as exc:
                if is_not_found_error(exc):
                    continue
                raise
            for nested in nested_entries:
                nested_name = nested.get("name", "")
                if not nested_name or nested_name in {".", ".."} or not nested.get("isDir"):
                    continue
                nested_path = f"{child_path.rstrip('/')}/{nested_name}"
                if not await self._looks_like_legacy_session_dir(nested_path):
                    continue
                if await self._legacy_session_visible(nested_path, ctx, owner_hint=name):
                    items.append((nested, f"viking://session/{nested_name}"))
        return items

    async def _session_root_items(
        self,
        uri: str,
        ctx: RequestContext,
    ) -> List[tuple[Dict[str, Any], str]]:
        primary_path = self._uri_to_path(uri, ctx=ctx)
        by_name: Dict[str, tuple[Dict[str, Any], str]] = {}
        try:
            for entry in await self._ls_entries(primary_path, ctx=ctx):
                name = entry.get("name", "")
                if not name or name in {".", ".."}:
                    continue
                by_name[name] = (entry, f"viking://session/{name}")
        except Exception as exc:
            if not is_not_found_error(exc):
                raise

        legacy_path = self._legacy_session_path(uri, ctx=ctx)
        for entry, entry_uri in await self._legacy_session_root_items(legacy_path, ctx):
            name = entry.get("name", "")
            if name and name not in by_name:
                by_name[name] = (entry, entry_uri)
        return list(by_name.values())

    async def _list_read_path_items(
        self,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> List[tuple[Dict[str, Any], str]]:
        real_ctx = self._ctx_or_default(ctx)
        if self._is_legacy_session_root_uri(uri):
            return await self._session_root_items(uri, real_ctx)

        primary_path = self._uri_to_path(uri, ctx=ctx)
        merge_paths = self._is_legacy_session_uri(uri)
        found_path = False
        last_not_found: Optional[Exception] = None
        by_uri: Dict[str, tuple[Dict[str, Any], str]] = {}

        for path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, path, primary_path, real_ctx):
                continue
            try:
                entries = await self._ls_entries(path, ctx=ctx)
            except Exception as exc:
                if is_not_found_error(exc):
                    last_not_found = exc
                    continue
                raise

            found_path = True
            for entry in entries:
                entry_uri = self._alias_uri_for_path(
                    request_uri=uri,
                    base_path=path,
                    entry_path=f"{path.rstrip('/')}/{entry.get('name', '')}",
                    ctx=ctx,
                )
                by_uri.setdefault(entry_uri, (entry, entry_uri))
            if not merge_paths:
                break

        if found_path:
            return list(by_uri.values())
        raise NotFoundError(uri, "directory") from last_not_found

    _ROOT_PATH = "/local"

    async def _ls_entries(
        self, path: str, ctx: Optional[RequestContext] = None
    ) -> List[Dict[str, Any]]:
        """List directory entries, filtering out internal directories.

        At account root (/local/{account}), uses LISTABLE_SCOPES whitelist.
        At other levels, uses the shared storage internal-name blacklist.
        """
        entries = await self._async_agfs.ls(path)
        parts = [p for p in path.strip("/").split("/") if p]
        if len(parts) == 2 and parts[0] == "local":
            return [e for e in entries if e.get("name") in VikingURI.LISTABLE_SCOPES]
        return [e for e in entries if e.get("name") not in STORAGE_INTERNAL_ENTRY_NAMES]

    def _path_to_uri(self, path: str, ctx: Optional[RequestContext] = None) -> str:
        """/local/{account}/... -> viking://...

        Pure prefix replacement: strips /local/{account_id}/ and prepends viking://.
        No implicit space stripping.
        """
        if path.startswith("viking://"):
            return path
        elif path.startswith("/local/"):
            inner = path[7:].strip("/")
            if not inner:
                return "viking://"
            real_ctx = self._ctx_or_default(ctx)
            parts = [p for p in inner.split("/") if p]
            if parts and parts[0] == real_ctx.account_id:
                parts = parts[1:]
            if not parts:
                return "viking://"
            return f"viking://{'/'.join(parts)}"
        elif path.startswith("/"):
            return f"viking:/{path}"
        else:
            return f"viking://{path}"

    def _looks_like_legacy_temp_leaf(self, value: str) -> bool:
        return bool(re.match(r"^\d{8}_[0-9a-f]{6}$", value or ""))

    def _is_legacy_temp_uri_parts(self, parts: List[str]) -> bool:
        if len(parts) < 2 or parts[0] != "temp" or not self._looks_like_legacy_temp_leaf(parts[1]):
            return False
        if len(parts) == 2:
            return True
        return not self._looks_like_legacy_temp_leaf(parts[2])

    def _is_accessible(self, uri: str, ctx: RequestContext) -> bool:
        """Check whether a URI is visible/accessible under current request context."""
        normalized_uri, parts = self._normalized_uri_parts(uri)
        if ctx.role == Role.ROOT:
            return True
        if is_hidden_by_actor_peer_view(normalized_uri, ctx):
            return False
        if not parts:
            return True
        if is_watch_task_control_uri(normalized_uri):
            return False

        scope = parts[0]
        if scope == "resources":
            return True
        if scope == "temp":
            if len(parts) == 1:
                return True
            if parts[1] == ctx.user.user_space_name():
                return True
            return self._is_legacy_temp_uri_parts(parts)
        if scope == "upload":
            return ctx.role == Role.ROOT
        if scope == "_system":
            return False
        if scope == "agent":
            # New format: agent/skills/..., agent/endpoints/... — globally readable (account scope)
            if len(parts) >= 2 and parts[1] in {"skills", "endpoints", "tools", "payments"}:
                return True
            # Old format: agent/{agent_id}/... — actor_peer_id match for read-only access
            if not ctx.actor_peer_id or len(parts) < 2:
                return True
            return parts[1] == ctx.actor_peer_id
        return namespace_is_accessible(normalized_uri, ctx)

    def _handle_agfs_read(self, result: Union[bytes, Any, None]) -> bytes:
        """Handle AGFSClient read return types consistently."""
        if isinstance(result, bytes):
            return result
        elif result is None:
            return b""
        elif hasattr(result, "content") and result.content is not None:
            return result.content
        else:
            # Try to convert to bytes
            try:
                return str(result).encode("utf-8")
            except Exception:
                return b""

    def _decode_bytes(self, data: bytes) -> str:
        """Robustly decode bytes to string."""
        if not data:
            return ""
        try:
            return data.decode("utf-8")
        except UnicodeDecodeError:
            try:
                # Try common encoding for Windows/legacy files in China
                return data.decode("gbk")
            except UnicodeDecodeError:
                try:
                    return data.decode("latin-1")
                except UnicodeDecodeError:
                    return data.decode("utf-8", errors="replace")

    def _handle_agfs_content(self, result: Union[bytes, Any, None]) -> str:
        """Handle AGFSClient content return types consistently."""
        if isinstance(result, bytes):
            return self._decode_bytes(result)
        elif hasattr(result, "content") and result.content is not None:
            return self._decode_bytes(result.content)
        elif result is None:
            return ""
        else:
            # Try to convert to string
            try:
                return str(result)
            except Exception:
                return ""

    # ========== Vector Sync Helper Methods ==========

    async def _collect_uris(
        self, path: str, recursive: bool, ctx: Optional[RequestContext] = None
    ) -> List[str]:
        """Recursively collect all URIs (for rm/mv), including directories."""
        uris = []

        async def _collect(p: str):
            try:
                for entry in await self._ls_entries(p, ctx=ctx):
                    name = entry.get("name", "")
                    if name in [".", ".."]:
                        continue
                    full_path = f"{p}/{name}".replace("//", "/")
                    if entry.get("isDir"):
                        uris.append(self._path_to_uri(full_path, ctx=ctx))
                        if recursive:
                            await _collect(full_path)
                    else:
                        uris.append(self._path_to_uri(full_path, ctx=ctx))
            except Exception:
                pass

        await _collect(path)
        return uris

    async def _delete_from_vector_store(
        self, uris: List[str], ctx: Optional[RequestContext] = None
    ) -> None:
        """Delete records with specified URIs from vector store.

        Uses tenant-safe URI deletion semantics from vector store.
        """
        vector_store = self._get_vector_store()
        if not vector_store:
            return
        real_ctx = self._ctx_or_default(ctx)

        try:
            await vector_store.delete_uris(real_ctx, uris)
            for uri in uris:
                logger.debug(f"[VikingFS] Deleted from vector store: {uri}")
        except Exception as e:
            logger.warning(f"[VikingFS] Failed to delete from vector store: {e}")

    async def _update_vector_store_uris(
        self,
        uris: List[str],
        old_base: str,
        new_base: str,
        ctx: Optional[RequestContext] = None,
        levels: Optional[List[int]] = None,
    ) -> None:
        """Update URIs in vector store (when moving files).

        Preserves vector data and updates URI-derived identifiers without regenerating embeddings.
        """
        vector_store = self._get_vector_store()
        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)

        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),
                    uri=uri,
                    new_uri=new_uri,
                    levels=levels,
                )
                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

    def _get_embedder(self) -> Any:
        """Get embedder instance."""
        return self.query_embedder

    # ========== Parent Directory Creation ==========

    async def _ensure_parent_dirs(self, path: str, ctx: Optional[RequestContext] = None) -> None:
        """Recursively create all parent directories."""
        try:
            await self._async_agfs.ensure_parent_dirs(path)
        except Exception as e:
            logger.debug(f"Failed to ensure parent directories for {path}: {e}")
            parent = path.rstrip("/").rsplit("/", 1)[0]
            await self._mkdir_path_with_parents(parent, ctx=ctx)

    async def _mkdir_path_with_parents(
        self, dir_path: str, ctx: Optional[RequestContext] = None
    ) -> None:
        """Create a directory path segment-by-segment using the same fs context."""
        parts = [part for part in dir_path.strip("/").split("/") if part]
        current = ""
        for part in parts:
            current = f"{current}/{part}"
            try:
                await self._async_agfs.mkdir(current)
            except Exception as e:
                message = str(e).lower()
                if "exist" in message or "already" in message:
                    continue
                logger.debug(f"Failed to create parent directory {current}: {e}")

    # ========== Relation Table Internal Methods ==========

    async def _read_relation_table(
        self, dir_path: str, ctx: Optional[RequestContext] = None
    ) -> List[RelationEntry]:
        """Read .relations.json."""
        table_path = f"{dir_path}/.relations.json"
        try:
            content = self._handle_agfs_read(await self._async_agfs.read(table_path))
            data = json.loads(content.decode("utf-8"))
        except FileNotFoundError:
            return []
        except Exception:
            # logger.warning(f"[VikingFS] Failed to read relation table {table_path}: {e}")
            return []

        entries = []
        # Compatible with old format (nested) and new format (flat)
        if isinstance(data, list):
            # New format: flat list
            for entry_data in data:
                entries.append(RelationEntry.from_dict(entry_data))
        elif isinstance(data, dict):
            # Old format: nested {namespace: {user: [entries]}}
            for _namespace, user_dict in data.items():
                for _user, entry_list in user_dict.items():
                    for entry_data in entry_list:
                        entries.append(RelationEntry.from_dict(entry_data))
        return entries

    async def _write_relation_table(
        self, dir_path: str, entries: List[RelationEntry], ctx: Optional[RequestContext] = None
    ) -> None:
        """Write .relations.json."""
        # Use flat list format
        data = [entry.to_dict() for entry in entries]

        content = json.dumps(data, ensure_ascii=False, indent=2)
        table_path = f"{dir_path}/.relations.json"
        if isinstance(content, str):
            content = content.encode("utf-8")

        await self._async_agfs.write(table_path, content)

    # ========== Batch Read (backward compatible) ==========

    async def read_batch(
        self, uris: List[str], level: str = "l0", ctx: Optional[RequestContext] = None
    ) -> Dict[str, str]:
        """Batch read content from multiple URIs."""
        results = {}
        for uri in uris:
            try:
                content = ""
                if level == "l0":
                    content = await self.abstract(uri, ctx=ctx)
                elif level == "l1":
                    content = await self.overview(uri, ctx=ctx)
                results[uri] = content
            except Exception:
                pass
        return results

    # ========== Other Preserved Methods ==========

    async def write_file(
        self,
        uri: str,
        content: Union[str, bytes],
        ctx: Optional[RequestContext] = None,
        lock_handle: Optional["LockHandle"] = None,
    ) -> None:
        """Write file directly."""
        self._ensure_mutable_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        await self._ensure_parent_dirs(path, ctx=ctx)

        if isinstance(content, str):
            content = content.encode("utf-8")

        await self._run_with_encrypted_write_lock(
            path,
            lambda: self._async_agfs.write(path, content),
            lock_handle=lock_handle,
        )

    async def read_file(
        self,
        uri: str,
        offset: int = 0,
        limit: int = -1,
        ctx: Optional[RequestContext] = None,
    ) -> str:
        """Read single file, optionally sliced by line range.

        Args:
            uri: Viking URI
            offset: Starting line number (0-indexed). Default 0.
            limit: Number of lines to read. -1 means read to end. Default -1.

        Raises:
            FileNotFoundError: If the file does not exist.
        """
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        # Verify the file exists before reading, because AGFS read returns
        # empty bytes for non-existent files instead of raising an error.
        last_not_found: Optional[Exception] = None
        for path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, path, primary_path, real_ctx):
                continue
            try:
                stat = await self._async_agfs.stat(path)
                break
            except Exception as exc:
                if is_not_found_error(exc):
                    last_not_found = exc
                    continue
                raise
        else:
            raise NotFoundError(uri, "file") from last_not_found
        if isinstance(stat, dict) and stat.get("isDir", False):
            raise InvalidArgumentError(
                f"Cannot read directory as file: {uri}",
                details={"resource": uri, "expected": "file", "actual": "directory"},
            )
        try:
            content = await self._async_agfs.read(path)
            if isinstance(content, bytes):
                raw = content
            elif content is not None and hasattr(content, "content"):
                raw = content.content
            else:
                raw = b""

            text = self._decode_bytes(raw)
        except Exception:
            raise NotFoundError(uri, "file")

        if offset == 0 and limit == -1:
            return text
        lines = text.splitlines(keepends=True)
        sliced = lines[offset:] if limit == -1 else lines[offset : offset + limit]
        return "".join(sliced)

    async def read_file_bytes(
        self,
        uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> bytes:
        """Read single binary file."""
        self._ensure_access(uri, ctx)
        real_ctx = self._ctx_or_default(ctx)
        primary_path = self._uri_to_path(uri, ctx=ctx)
        last_not_found: Optional[Exception] = None
        for path in self._read_paths(uri, ctx=ctx):
            if not await self._read_path_visible(uri, path, primary_path, real_ctx):
                continue
            try:
                stat = await self._async_agfs.stat(path)
                break
            except Exception as exc:
                if is_not_found_error(exc):
                    last_not_found = exc
                    continue
                raise
        else:
            raise NotFoundError(uri, "file") from last_not_found
        if isinstance(stat, dict) and stat.get("isDir", False):
            raise InvalidArgumentError(
                f"Cannot read directory as file: {uri}",
                details={"resource": uri, "expected": "file", "actual": "directory"},
            )
        try:
            raw = self._handle_agfs_read(await self._async_agfs.read(path))
            return raw
        except Exception:
            raise NotFoundError(uri, "file")

    async def write_file_bytes(
        self,
        uri: str,
        content: bytes,
        ctx: Optional[RequestContext] = None,
        lock_handle: Optional["LockHandle"] = None,
    ) -> None:
        """Write single binary file."""
        self._ensure_mutable_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)
        await self._ensure_parent_dirs(path, ctx=ctx)

        await self._run_with_encrypted_write_lock(
            path,
            lambda: self._async_agfs.write(path, content),
            lock_handle=lock_handle,
        )

    async def append_file(
        self,
        uri: str,
        content: str,
        ctx: Optional[RequestContext] = None,
        lock_handle: Optional["LockHandle"] = None,
    ) -> None:
        """Append content to file."""
        self._ensure_mutable_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)

        try:
            await self._ensure_parent_dirs(path, ctx=ctx)

            async def _append_under_lock() -> None:
                """Read old content and rewrite the whole file under lock to avoid lost updates."""
                existing = ""
                try:
                    existing_bytes = self._handle_agfs_read(await self._async_agfs.read(path))
                    existing = self._decode_bytes(existing_bytes)
                except FileNotFoundError:
                    pass
                except AGFSHTTPError as e:
                    if e.status_code != 404:
                        raise
                except AGFSClientError:
                    raise

                final_content = (existing + content).encode("utf-8")
                await self._async_agfs.write(path, final_content)

            await self._run_with_encrypted_write_lock(
                path,
                _append_under_lock,
                lock_handle=lock_handle,
            )

        except Exception as e:
            logger.error(f"[VikingFS] Failed to append to file {uri}: {e}")
            raise IOError(f"Failed to append to file {uri}: {e}")

    async def ls(
        self,
        uri: str,
        output: str = "original",
        abs_limit: int = 256,
        show_all_hidden: bool = False,
        node_limit: int = 1000,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """
        List directory contents (URI version).

        Args:
            uri: Viking URI
            output: str = "original"
            abs_limit: int = 256
            show_all_hidden: bool = False (list all hidden files, like -a)
            node_limit: int = 1000 (maximum number of nodes to list)

        output="original"
        [{'name': '.abstract.md', 'size': 100, 'mode': 420, 'modTime': '2026-02-11T16:52:16.256334192+08:00', 'isDir': False, 'meta': {'Name': 'localfs', 'Type': 'local', 'Content': None}, 'uri': 'viking://resources/.abstract.md'}]

        output="agent"
        [{'name': '.abstract.md', 'size': 100, 'modTime': '2026-02-11T08:52:16.256Z', 'isDir': False, 'uri': 'viking://resources/.abstract.md', 'abstract': "..."}]
        """
        self._ensure_access(uri, ctx)
        if output == "original":
            return await self._ls_original(uri, show_all_hidden, node_limit, ctx=ctx)
        elif output == "agent":
            return await self._ls_agent(uri, abs_limit, show_all_hidden, node_limit, ctx=ctx)
        else:
            raise ValueError(f"Invalid output format: {output}")

    async def _ls_agent(
        self,
        uri: str,
        abs_limit: int,
        show_all_hidden: bool,
        node_limit: int = 1000,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """List directory contents (URI version)."""
        real_ctx = self._ctx_or_default(ctx)
        entry_items = await self._list_read_path_items(uri, ctx=ctx)
        # basic info
        fallback_time = datetime.now(timezone.utc)
        all_entries = []
        for entry, entry_uri in entry_items:
            if len(all_entries) >= node_limit:
                break
            name = entry.get("name", "")
            raw_time = entry.get("modTime", "")
            parsed_time = fallback_time
            if isinstance(raw_time, (int, float)):
                parsed_time = datetime.fromtimestamp(raw_time, tz=timezone.utc)
            elif raw_time:
                if len(raw_time) > 26 and "+" in raw_time:
                    parts = raw_time.split("+")
                    raw_time = parts[0][:26] + "+" + parts[1]
                parsed_time = parse_iso_datetime(raw_time)
            elif isinstance(entry.get("mtime"), (int, float)):
                parsed_time = datetime.fromtimestamp(entry["mtime"], tz=timezone.utc)
            is_dir = entry.get("isDir", False)
            new_entry = {
                "uri": entry_uri,
                "size": 0 if is_dir else entry.get("size", 0),
                "isDir": is_dir,
                "modTime": format_iso8601(parsed_time),
            }
            if not self._is_accessible(new_entry["uri"], real_ctx):
                continue
            if is_dir:
                all_entries.append(new_entry)
            elif not name.startswith("."):
                all_entries.append(new_entry)
            elif show_all_hidden:
                all_entries.append(new_entry)
        await self._batch_fetch_abstracts(all_entries, abs_limit, ctx=ctx)
        return all_entries

    async def _ls_original(
        self,
        uri: str,
        show_all_hidden: bool = False,
        node_limit: int = 1000,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """List directory contents (URI version)."""
        real_ctx = self._ctx_or_default(ctx)
        try:
            entry_items = await self._list_read_path_items(uri, ctx=ctx)
            # AGFS returns read-only structure, need to create new dict
            all_entries = []
            for entry, entry_uri in entry_items:
                if len(all_entries) >= node_limit:
                    break
                name = entry.get("name", "")
                new_entry = dict(entry)  # Copy original data
                new_entry["uri"] = entry_uri
                if not self._is_accessible(new_entry["uri"], real_ctx):
                    continue
                if entry.get("isDir"):
                    all_entries.append(new_entry)
                elif not name.startswith("."):
                    all_entries.append(new_entry)
                elif show_all_hidden:
                    all_entries.append(new_entry)
            return all_entries
        except Exception:
            raise NotFoundError(uri, "directory")

    async def move_file(
        self,
        from_uri: str,
        to_uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Move file."""
        self._ensure_mutable_access(from_uri, ctx)
        self._ensure_mutable_access(to_uri, ctx)
        from_path = self._uri_to_path(from_uri, ctx=ctx)

        await self._copy_file_through_vikingfs(from_uri, to_uri, ctx=ctx)
        await self._async_agfs.rm(from_path)

    # ========== Temp File Operations (backward compatible) ==========

    def create_temp_uri(self, ctx: Optional[RequestContext] = None) -> str:
        """Create a temp directory URI.

        - explicit ctx or bound request context -> user-scoped temp URI
        - no explicit/bound context -> legacy temp URI shape for backward compatibility
        """
        real_ctx = ctx if ctx is not None else self._bound_ctx.get()
        if real_ctx is None:
            return VikingURI.create_temp_uri()
        return VikingURI.create_temp_uri(space=real_ctx.user.user_space_name())

    async def persist_temp_tree(
        self,
        temp_uri: str,
        target_uri: str,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Persist an already-encrypted temp tree without rewriting file bytes."""
        self._ensure_access(temp_uri, ctx)
        self._ensure_mutable_access(target_uri, ctx)
        src_path = self._uri_to_path(temp_uri, ctx=ctx)
        dst_path = self._uri_to_path(target_uri, ctx=ctx)
        await self._ensure_parent_dirs(dst_path, ctx=ctx)
        await self._async_agfs.cp(
            src_path,
            dst_path,
            recursive=True,
            fs_ctx={"account_id": self._ctx_or_default(ctx).account_id},
        )

    async def delete_temp(self, temp_uri: str, ctx: Optional[RequestContext] = None) -> None:
        """Delete temp directory and its contents."""
        self._ensure_mutable_access(temp_uri, ctx)
        path = self._uri_to_path(temp_uri, ctx=ctx)
        try:
            for entry in await self._ls_entries(path, ctx=ctx):
                name = entry.get("name", "")
                if name in [".", ".."]:
                    continue
                entry_path = f"{path}/{name}"
                if entry.get("isDir"):
                    await self.delete_temp(f"{temp_uri}/{name}", ctx=ctx)
                else:
                    await self._async_agfs.rm(entry_path)
            await self._async_agfs.rm(path)
        except Exception as e:
            logger.warning(f"[VikingFS] Failed to delete temp {temp_uri}: {e}")

    async def get_relations(self, uri: str, ctx: Optional[RequestContext] = None) -> List[str]:
        """Get all related URIs (backward compatible)."""
        entries = await self.get_relation_table(uri, ctx=ctx)
        real_ctx = self._ctx_or_default(ctx)
        all_uris = []
        for entry in entries:
            for related in entry.uris:
                if self._is_accessible(related, real_ctx):
                    all_uris.append(related)
        return all_uris

    async def get_relations_with_content(
        self,
        uri: str,
        include_l0: bool = True,
        include_l1: bool = False,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """Get related URIs and their content (backward compatible)."""
        relation_uris = await self.get_relations(uri, ctx=ctx)
        if not relation_uris:
            return []

        results = []
        abstracts = {}
        overviews = {}
        if include_l0:
            abstracts = await self.read_batch(relation_uris, level="l0", ctx=ctx)
        if include_l1:
            overviews = await self.read_batch(relation_uris, level="l1", ctx=ctx)

        for rel_uri in relation_uris:
            info = {"uri": rel_uri}
            if include_l0:
                info["abstract"] = abstracts.get(rel_uri, "")
            if include_l1:
                info["overview"] = overviews.get(rel_uri, "")
            results.append(info)

        return results

    async def write_context(
        self,
        uri: str,
        content: Union[str, bytes] = "",
        abstract: str = "",
        overview: str = "",
        content_filename: str = "content.md",
        is_leaf: bool = False,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Write context to AGFS (L0/L1/L2)."""

        self._ensure_mutable_access(uri, ctx)
        path = self._uri_to_path(uri, ctx=ctx)

        try:
            await self._ensure_parent_dirs(path, ctx=ctx)
            try:
                await self._async_agfs.mkdir(path)
            except Exception as e:
                if "exist" not in str(e).lower():
                    raise

            if content:
                content_uri = f"{uri}/{content_filename}"
                await self.write_file(content_uri, content, ctx=ctx)

            if abstract:
                abstract_uri = f"{uri}/.abstract.md"
                await self.write_file(abstract_uri, abstract, ctx=ctx)

            if overview:
                overview_uri = f"{uri}/.overview.md"
                await self.write_file(overview_uri, overview, ctx=ctx)

        except Exception as e:
            logger.error(f"[VikingFS] Failed to write {uri}: {e}")
            raise IOError(f"Failed to write {uri}: {e}")

    # ------------------------------------------------------------------
    # Git version control (commit / restore / show / log)
    # ------------------------------------------------------------------

    # First path segments that the Rust git enumerate.rs prunes from snapshots,
    # plus the runtime lock name. Mirrors INTERNAL_FIRST_SEGMENTS in
    # crates/ragfs/src/git/enumerate.rs and VikingFS._INTERNAL_NAMES so that
    # callers fail fast in Python with a clear error rather than passing a
    # path that the Rust side will silently drop.
    _GIT_INTERNAL_FIRST_SEGMENTS = frozenset(
        {
            "_system",
            "tasks",
            "temp",
            "queue",
            "upload",
            ".path.ovlock",
        }
    )

    _DEFAULT_GIT_AUTHOR_NAME = "viking-bot"
    _DEFAULT_GIT_AUTHOR_EMAIL = "bot@viking.local"

    _OVGITIGNORE_TREE_PATH = ".ovgitignore"

    # Must stay in sync with `OVGITIGNORE_MAX_BYTES` in the Rust layer
    # (`crates/ragfs/src/git/ignore.rs`); enforced here at write time so a bad
    # file can never be persisted only to poison every later commit.
    _OVGITIGNORE_MAX_BYTES = 64 * 1024

    def _gitignore_agfs_path(self, ctx: Optional[RequestContext] = None) -> str:
        real_ctx = self._ctx_or_default(ctx)
        return f"/local/{real_ctx.account_id}/{self._OVGITIGNORE_TREE_PATH}"

    def _uri_to_tree_path(self, uri: str, ctx: Optional[RequestContext] = None) -> str:
        """Convert a viking:// URI to an account-relative git tree path.

        ``viking://resources/proj_a/docs/a.md`` -> ``resources/proj_a/docs/a.md``.

        Pure prefix stripping: removes the ``viking://`` scheme and any
        ``/local/{account}/`` segment. Internal scopes that the Rust git layer
        would prune (`_system`, `tasks`, `temp`, `queue`, `upload`) and the
        runtime lock name (`.path.ovlock`) are rejected with ``ValueError``
        — passing them through silently would result in a no-op commit and
        confuse callers.
        """
        real_ctx = self._ctx_or_default(ctx)
        canonical = canonicalize_uri(uri, real_ctx)
        _, parts = self._normalized_uri_parts(canonical)
        if not parts:
            raise ValueError(f"git tree path cannot be the account root: {uri!r}")
        first = parts[0]
        if first in self._GIT_INTERNAL_FIRST_SEGMENTS:
            raise ValueError(f"git tree path rejects internal scope/segment {first!r}: {uri!r}")
        return "/".join(parts)

    def _tree_path_to_uri(self, tree_path: str) -> str:
        """Convert an account-relative git tree path to a viking:// URI.

        Inverse of :py:meth:`_uri_to_tree_path` (without context canonicalization).
        """
        cleaned = tree_path.strip("/")
        if not cleaned:
            raise ValueError("tree path must not be empty")
        return f"viking://{cleaned}"

    _DIR_MARKER_LEVELS = {
        ".abstract.md": ContextLevel.ABSTRACT,
        ".overview.md": ContextLevel.OVERVIEW,
    }
    _NO_VECTOR_DERIVED = frozenset({".relations.json", ".ovgitignore"})

    def _classify_restore_path(self, tree_path: str, *, deleted: bool) -> Optional[tuple]:
        """Classify a restore-affected tree path into a vector maintenance task.

        Returns a ``(op, uri, level)`` triple, or ``None`` when the path has no
        vector side-effect:

        - ``dir/.abstract.md`` / ``dir/.overview.md`` → recompute (write) or
          delete (removal) ONLY that directory's L0/L1 vector:
          ``("reindex_marker"|"delete", dir_uri, ABSTRACT|OVERVIEW)``.
        - ``.relations.json`` → ``None`` (not a vector text source).
        - anything else (a source file) → reindex (write) or delete (removal)
          its DETAIL vector:
          ``("reindex_file", file_uri, DETAIL)`` / ``("delete", file_uri, DETAIL)``.

        ``None`` is also returned for a directory marker at the account root
        (no parent directory to scope an L0/L1 vector to).
        """
        parent, _, name = tree_path.rpartition("/")
        if name in self._NO_VECTOR_DERIVED:
            return None
        level = self._DIR_MARKER_LEVELS.get(name)
        if level is not None:
            if not parent:
                return None
            dir_uri = self._tree_path_to_uri(parent)
            op = "delete" if deleted else "reindex_marker"
            return (op, dir_uri, level)
        # Source file.
        file_uri = self._tree_path_to_uri(tree_path)
        op = "delete" if deleted else "reindex_file"
        return (op, file_uri, ContextLevel.DETAIL)

    async def get_gitignore(self, ctx: Optional[RequestContext] = None) -> str:
        """Return the account-level .ovgitignore content, or an empty string if absent.

        Raises ``AGFSInvalidOperationError`` if the stored content is not valid
        UTF-8, mirroring the Rust layer's reject-non-UTF-8 behavior at commit
        time rather than leaking a raw ``UnicodeDecodeError``.
        """
        path = self._gitignore_agfs_path(ctx)
        try:
            raw = await self._async_agfs.read(path, 0, -1)
        except Exception as exc:
            if is_not_found_error(exc):
                return ""
            mapped = map_exception(exc, resource=path)
            if mapped is not None:
                raise mapped from exc
            raise
        if isinstance(raw, bytes):
            data = raw
        elif raw is not None and hasattr(raw, "content"):
            data = raw.content
        else:
            data = b""
        try:
            return data.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise AGFSInvalidOperationError(
                f"invalid ignore file {path}: must be UTF-8: {exc}"
            ) from exc

    async def set_gitignore(
        self,
        content: str,
        ctx: Optional[RequestContext] = None,
    ) -> None:
        """Write the account-level .ovgitignore control file without semantic indexing.

        Validates the size limit up front so a too-large file can never be
        persisted only to make every later ``commit`` fail. Syntax (negation,
        escaping) is still validated at commit time by the Rust layer.
        """
        if not isinstance(content, str):
            raise TypeError("content must be a string")
        path = self._gitignore_agfs_path(ctx)
        data = content.encode("utf-8")
        if len(data) > self._OVGITIGNORE_MAX_BYTES:
            raise AGFSInvalidOperationError(
                f"ignore file too large: {path} is {len(data)} bytes, "
                f"limit {self._OVGITIGNORE_MAX_BYTES} bytes"
            )
        await self._ensure_parent_dirs(path, ctx=ctx)
        await self._async_agfs.write(path, data)

    async def delete_gitignore(self, ctx: Optional[RequestContext] = None) -> None:
        """Delete the account-level .ovgitignore control file. Missing is success."""
        path = self._gitignore_agfs_path(ctx)
        try:
            await self._async_agfs.rm(path, recursive=False)
        except Exception as exc:
            if is_not_found_error(exc):
                return
            mapped = map_exception(exc, resource=path)
            if mapped is not None:
                raise mapped from exc
            raise

    async def commit(
        self,
        *,
        message: str,
        paths: Optional[List[str]] = None,
        branch: str = "main",
        author_name: Optional[str] = None,
        author_email: Optional[str] = None,
        ctx: Optional[RequestContext] = None,
    ) -> Dict[str, Any]:
        """Create a git snapshot of the account's tree.

        Args:
            message: Commit message.
            paths: Optional list of ``viking://`` URIs to scope the commit to;
                entries may be files or directories. Directories are expanded
                recursively with the snapshot pruning rules applied. ``None``
                (default) enumerates the whole account tree. An empty list is
                forwarded as an explicit empty path list (no-op commit). A
                path that exists in neither the VFS nor the previous snapshot
                logs a warning and is treated as a no-op deletion.
            branch: Branch to advance. Defaults to ``"main"``.
            author_name / author_email: Override the default bot author.
            ctx: Request context (provides ``account_id``).

        Returns:
            Dict with ``result`` (``"created"`` / ``"noop"``) and ``commit_oid``.
            When ``result == "created"`` it also includes ``changed`` (paths
            added/updated/removed). Both variants include ``ignored``: the
            number of candidate paths skipped by the account ``.ovgitignore``
            rules (system pruning is not counted). Noop omits ``changed``.
        """
        real_ctx = self._ctx_or_default(ctx)
        account = real_ctx.account_id
        if paths is None:
            tree_paths: Optional[List[str]] = None
        else:
            tree_paths = [self._uri_to_tree_path(p, ctx=real_ctx) for p in paths]
        return await self._async_agfs.run(
            "git_commit",
            account=account,
            branch=branch,
            message=message,
            paths=tree_paths,
            author_name=author_name or self._DEFAULT_GIT_AUTHOR_NAME,
            author_email=author_email or self._DEFAULT_GIT_AUTHOR_EMAIL,
        )

    async def restore(
        self,
        *,
        project_dir: Optional[str] = None,
        source_commit: str,
        branch: str = "main",
        dry_run: bool = False,
        message: Optional[str] = None,
        author_name: Optional[str] = None,
        author_email: Optional[str] = None,
        ctx: Optional[RequestContext] = None,
    ) -> Dict[str, Any]:
        """Restore a project subtree to the state at ``source_commit``.

        Generates a new commit whose parent is the current HEAD (not the
        source commit) and writes the diff through the VFS. Paths outside
        ``project_dir`` are left untouched.

        Args:
            project_dir: ``viking://`` URI of the subtree, e.g.
                ``"viking://resources/proj_a"``. May also be passed as a
                short form like ``"resources/proj_a"``. Trailing slashes are
                stripped. Internal scopes are rejected with ``ValueError``.
                If None, restore the entire tree.
            source_commit: 40-hex OID, branch name, or full ref path.
            branch: Branch to advance. Defaults to ``"main"``.
            dry_run: If True, returns the planned diff without writing.
            message: Optional commit message; defaults to a generated string.
            author_name / author_email: Override the default bot author.
            ctx: Request context (provides ``account_id``).

        Returns:
            Dict containing ``result`` (``"applied"`` / ``"noop"`` / ``"dry_run"``)
            and corresponding oid / diff fields. When an ``Applied`` result has
            vector side-effects, a ``task_id`` is included for polling the
            background reindex via ``GET /api/v1/tasks/{task_id}``.

        After an ``Applied`` result, this method schedules background vector
        maintenance for the affected paths via :class:`ReindexExecutor`:
        directory markers (``.abstract.md`` / ``.overview.md``) recompute or
        delete only that directory's L0/L1 vector; source files reindex (write)
        or delete (removal) their DETAIL vector. ``.relations.json`` has no
        vector side-effect. The rebuild is tracked as a single task
        (``snapshot_restore_reindex``); per-path failures are logged and do not
        block the return value. The task reaches ``completed`` only after each
        affected vector has been written to (or deleted from) the index, so
        polling ``task_id`` to ``completed`` guarantees subsequent ``find``
        reads see the restored state.
        """
        real_ctx = self._ctx_or_default(ctx)
        account = real_ctx.account_id
        tree_dir: Optional[str]
        if project_dir is None:
            tree_dir = None
        else:
            tree_dir = self._uri_to_tree_path(project_dir, ctx=real_ctx).rstrip("/")
            if not tree_dir:
                raise ValueError(f"project_dir must not be empty: {project_dir!r}")
        # Build kwargs dynamically, only include project_dir if it's not None
        kwargs = {
            "account": account,
            "branch": branch,
            "source_commit": source_commit,
            "dry_run": dry_run,
            "message": message,
            "author_name": author_name or self._DEFAULT_GIT_AUTHOR_NAME,
            "author_email": author_email or self._DEFAULT_GIT_AUTHOR_EMAIL,
        }
        if tree_dir is not None:
            kwargs["project_dir"] = tree_dir
        # dry_run only computes the diff; it never writes the VFS, so it needs
        # no lock.
        if dry_run:
            return await self._async_agfs.run("git_restore", **kwargs)

        from openviking.pyagfs.exceptions import GitRestoreWritebackPartialError
        from openviking.storage.errors import LockAcquisitionError, ResourceBusyError
        from openviking.storage.transaction import LockContext, get_lock_manager

        # Serialize the writeback against concurrent VFS mutations on the same
        # subtree. A scoped restore tree-locks project_dir; a full restore
        # (project_dir is None) locks the account root so only one restore runs
        # per account at a time. The lock covers only the writeback — the
        # background reindex is scheduled after release, otherwise its per-path
        # child locks would conflict with this tree lock.
        lock_path = (
            self._uri_to_path(project_dir, ctx=real_ctx)
            if project_dir is not None
            else f"/local/{account}"
        )
        partial_exc: Optional[GitRestoreWritebackPartialError] = None
        try:
            async with LockContext(get_lock_manager(), [lock_path], lock_mode="tree"):
                try:
                    result = await self._async_agfs.run("git_restore", **kwargs)
                except GitRestoreWritebackPartialError as exc:
                    # The ref already advanced — capture the exception and
                    # finish the lock scope cleanly. Reindex scheduling and
                    # re-raise happen below, outside the tree lock.
                    partial_exc = exc
                    result = None
        except LockAcquisitionError:
            raise ResourceBusyError(
                f"Resource is being processed: {project_dir or '*'}",
                uri=project_dir or "*",
            )

        if partial_exc is not None:
            # HEAD has moved forward but some VFS writes/deletes failed.
            # Still schedule reindex for the paths that *did* reach the VFS so
            # the vector index doesn't stay stale, then re-raise so the caller
            # learns about the partial failure (and can inspect failed_writes
            # / failed_deletes / task_id on the exception).
            try:
                partial_exc.task_id = await self._schedule_restore_reindex_for_paths(
                    written_paths=partial_exc.written_paths,
                    deleted_paths=partial_exc.deleted_paths,
                    project_dir=project_dir,
                    real_ctx=real_ctx,
                )
            except Exception:
                logger.exception(
                    "[VikingFS] git restore partial: reindex scheduling failed; "
                    "HEAD advanced but reindex was not queued"
                )
            raise partial_exc

        if result.get("result") != "applied":
            return result

        written = list(result.get("written_paths") or [])
        deleted = list(result.get("deleted_paths") or [])
        if written or deleted:
            try:
                task_id = await self._schedule_restore_reindex_for_paths(
                    written_paths=written,
                    deleted_paths=deleted,
                    project_dir=project_dir,
                    real_ctx=real_ctx,
                )
                if task_id is not None:
                    result["task_id"] = task_id
            except Exception:
                logger.exception(
                    "[VikingFS] git restore reindex task creation failed; "
                    "falling back to fire-and-forget rebuild"
                )
                self._schedule_vector_rebuild(written=written, deleted=deleted, ctx=real_ctx)
        return result

    async def _schedule_restore_reindex_for_paths(
        self,
        *,
        written_paths: List[str],
        deleted_paths: List[str],
        project_dir: Optional[str],
        real_ctx: RequestContext,
    ) -> Optional[str]:
        """Classify ``written``/``deleted`` paths into vector tasks and queue
        them as a single tracked background rebuild. Returns the task id, or
        ``None`` if there is nothing to do.

        Shared by the applied path and the partial-writeback recovery path —
        both schedule reindex for paths that actually reached the VFS.
        """
        if not written_paths and not deleted_paths:
            return None
        tasks = self._collect_restore_vector_tasks(written_paths, deleted_paths)
        if not tasks:
            return None

        from openviking.service.task_tracker import get_task_tracker

        tracker = get_task_tracker()
        task = await tracker.create(
            "snapshot_restore_reindex",
            resource_id=project_dir or "*",
            account_id=real_ctx.account_id,
            user_id=real_ctx.user.user_id,
        )
        await tracker.update_stage(
            task.task_id,
            "queued",
            account_id=real_ctx.account_id,
            user_id=real_ctx.user.user_id,
        )
        background = asyncio.create_task(
            self._run_restore_rebuild_tracked(task.task_id, tasks, real_ctx),
            name=f"vikingfs-git-restore-reindex:{task.task_id}",
        )
        self._background_tasks.add(background)
        background.add_done_callback(self._background_tasks.discard)
        return task.task_id

    async def show(
        self,
        target_ref: str,
        *,
        path: Optional[str] = None,
        ctx: Optional[RequestContext] = None,
    ) -> Union[Dict[str, Any], bytes]:
        """Read a commit's metadata or a single blob.

        ``path=None`` returns the commit metadata dict (oid, tree, parents,
        author, committer, message). ``path=str`` returns the blob bytes
        directly, stripping the oid/size envelope returned by the binding.
        """
        real_ctx = self._ctx_or_default(ctx)
        account = real_ctx.account_id
        tree_path = self._uri_to_tree_path(path, ctx=real_ctx) if path else None
        resp = await self._async_agfs.run(
            "git_show",
            account=account,
            target_ref=target_ref,
            path=tree_path,
        )
        if path is not None and isinstance(resp, dict) and "bytes" in resp:
            return resp["bytes"]
        return resp

    async def show_blob_raw(
        self,
        target_ref: str,
        *,
        path: str,
        ctx: Optional[RequestContext] = None,
    ) -> Dict[str, Any]:
        """Like ``show(target_ref, path=...)`` but returns the full envelope.

        Returns ``{"oid": str, "size": int, "bytes": bytes}`` without
        stripping. Used by the HTTP snapshot router to populate
        ``X-Snapshot-Oid`` / ``X-Snapshot-Size`` response headers.
        """
        real_ctx = self._ctx_or_default(ctx)
        account = real_ctx.account_id
        tree_path = self._uri_to_tree_path(path, ctx=real_ctx)
        resp = await self._async_agfs.run(
            "git_show",
            account=account,
            target_ref=target_ref,
            path=tree_path,
        )
        if not isinstance(resp, dict) or "bytes" not in resp:
            raise TypeError(
                f"git_show returned unexpected shape for blob path: {type(resp).__name__}"
            )
        return resp

    async def log(
        self,
        *,
        branch: str = "main",
        limit: int = 20,
        ctx: Optional[RequestContext] = None,
    ) -> List[Dict[str, Any]]:
        """Walk back from ``branch``'s HEAD along ``parents[0]`` up to ``limit`` commits.

        Returns a list of commit metadata dicts (same shape as ``show(target_ref)``).
        """
        if limit <= 0:
            return []
        real_ctx = self._ctx_or_default(ctx)
        account = real_ctx.account_id
        head = await self._async_agfs.run(
            "git_show",
            account=account,
            target_ref=branch,
            path=None,
        )
        results: List[Dict[str, Any]] = [head]
        parents = head.get("parents") or []
        while parents and len(results) < limit:
            parent_oid = parents[0]
            commit = await self._async_agfs.run(
                "git_show",
                account=account,
                target_ref=parent_oid,
                path=None,
            )
            results.append(commit)
            parents = commit.get("parents") or []
        return results

    def _collect_restore_vector_tasks(
        self,
        written: List[str],
        deleted: List[str],
    ) -> set[tuple]:
        """Classify restore-affected paths into deduplicated ``(op, uri, level)`` tasks.

        Tasks are deduplicated on the exact ``(op, uri, level)`` key (no ancestor
        subsumption — each change is handled independently because no operation
        recurses into descendants).
        """
        tasks: set[tuple] = set()
        for tree_path in written:
            try:
                task = self._classify_restore_path(tree_path, deleted=False)
            except ValueError:
                continue
            if task is not None:
                tasks.add(task)
        for tree_path in deleted:
            try:
                task = self._classify_restore_path(tree_path, deleted=True)
            except ValueError:
                continue
            if task is not None:
                tasks.add(task)
        return tasks

    def _schedule_vector_rebuild(
        self,
        *,
        written: List[str],
        deleted: List[str],
        ctx: RequestContext,
    ) -> None:
        """Fire-and-forget precise vector maintenance for a git restore.

        Each affected path is classified by :py:meth:`_classify_restore_path`
        into a ``(op, uri, level)`` task and scheduled independently:

        - ``reindex_marker`` — recompute only a directory's L0/L1 vector.
        - ``reindex_file`` — recompute only a source file's DETAIL vector
          (non-recursive ``execute(mode="vectors_only")``).
        - ``delete`` — remove only the ``(uri, level)`` vector (directory
          marker removal, or deleted source file).

        Failures are logged and never propagate.
        """
        try:
            from openviking.service.reindex_executor import get_reindex_executor
        except Exception:
            logger.exception("[VikingFS] ReindexExecutor import failed; skipping rebuild")
            return

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            logger.warning("[VikingFS] git restore vector rebuild skipped: no running event loop")
            return

        tasks = self._collect_restore_vector_tasks(written, deleted)
        if not tasks:
            return

        executor = get_reindex_executor()
        for op, uri, level in tasks:
            loop.create_task(
                self._run_vector_rebuild(executor, op, uri, level, ctx),
                name=f"vikingfs-git-{op}:{uri}:{int(level)}",
            )

    async def _run_restore_rebuild_tracked(
        self,
        task_id: str,
        tasks: set[tuple],
        ctx: RequestContext,
    ) -> None:
        """Background worker driving a tracked restore vector rebuild.

        Runs all classified ``(op, uri, level)`` rebuild tasks concurrently and
        drives the task through start → complete/fail. Per-task failures are
        swallowed (and logged) inside :py:meth:`_run_vector_rebuild`, preserving
        the "failures do not block" semantics.
        """
        from openviking.service.task_tracker import get_task_tracker

        tracker = get_task_tracker()
        await tracker.start(
            task_id,
            account_id=ctx.account_id,
            user_id=ctx.user.user_id,
            stage="reindexing",
        )
        try:
            from openviking.service.reindex_executor import get_reindex_executor

            executor = get_reindex_executor()
            await asyncio.gather(
                *[
                    self._run_vector_rebuild(executor, op, uri, level, ctx)
                    for (op, uri, level) in tasks
                ]
            )
            await tracker.complete(
                task_id,
                {"status": "completed", "task_count": len(tasks)},
                account_id=ctx.account_id,
                user_id=ctx.user.user_id,
            )
        except Exception as exc:
            await tracker.fail(
                task_id,
                str(exc),
                account_id=ctx.account_id,
                user_id=ctx.user.user_id,
            )

    async def _run_vector_rebuild(
        self,
        executor: Any,
        op: str,
        uri: str,
        level: ContextLevel,
        ctx: RequestContext,
    ) -> None:
        """Wrapper coroutine: dispatch one vector task and swallow errors."""
        try:
            if op == "reindex_marker":
                await executor.reindex_directory_marker(dir_uri=uri, level=level, ctx=ctx)
            elif op == "reindex_file":
                await executor.execute(uri=uri, mode="vectors_only", wait=True, ctx=ctx)
            elif op == "delete":
                await executor.delete_uri_level(uri=uri, level=level, ctx=ctx)
            else:  # pragma: no cover - defensive
                logger.warning("[VikingFS] unknown vector rebuild op %r for %s", op, uri)
        except Exception:
            logger.exception("[VikingFS] git restore vector task %s failed for %s", op, uri)
