[1/69] src/remote_store/__init__.py (146 rows, definitions)

[2/69] src/remote_store/_async_to_sync_adapter.py (875 rows, definitions)
---
```python
@runtime_checkable
class AsyncBackendSyncAdapter(Backend):
    """Wraps an ``AsyncBackend`` as a synchronous ``Backend``.

    One private ``asyncio`` event loop per adapter instance, running on
    a dedicated daemon thread for the adapter's lifetime.  Sync methods
    submit coroutines via ``asyncio.run_coroutine_threadsafe`` and
    block on the returned ``concurrent.futures.Future``.

    Construction does not enter the wrapped backend's async context
    manager -- callers that need ``__aenter__`` semantics should use
    ``AsyncStore`` directly.

    Args:
        async_backend: The async backend instance to wrap.
    """
    def __init__(self, async_backend: AsyncBackend) -> None:
    @property
    def name(self) -> str:
        """Backend identifier, forwarded from the wrapped async backend."""
    @functools.cached_property
    def capabilities(self) -> CapabilitySet:
        """Capabilities reported by the wrapped backend, with stream-shape
        translation applied.

        ``SEEKABLE_READ`` is masked off unconditionally — the chunk-pull
        stream this adapter returns is forward-only.  ``USER_METADATA``
        and ``WRITE_RESULT_NATIVE`` are forwarded from the inner async
        backend unchanged; the ``write*()`` methods accept and forward
        ``metadata=``.
        """
    def to_key(self, native_path: str) -> str:
    def native_path(self, path: str) -> str:
    def resolve(self, path: str) -> ResolutionPlan:
    def unwrap(self, type_hint: type[T]) -> T:
        """Return a sync-safe native handle if the wrapped backend provides one.

        By default raises ``CapabilityNotSupported``
        because async-SDK handles are bound to the adapter's private event loop
        and cannot be used safely from the caller's thread.

        Wrapped backends that can expose a sync-safe handle should implement
        ``_SyncSafeHandleProvider`` and return it from
        ``_SyncSafeHandleProvider.sync_safe_unwrap``.

        Args:
            type_hint: The type of handle to retrieve; passed through to
                ``_SyncSafeHandleProvider.sync_safe_unwrap`` for backends
                that support the exemption.

        Returns:
            A sync-safe handle of the type requested via *type_hint*.

        Raises:
            CapabilityNotSupported: If the wrapped backend does not implement
                ``_SyncSafeHandleProvider``.

        Example:
            ```python
            class _SafeBackend(AsyncBackend, _SyncSafeHandleProvider):
                def sync_safe_unwrap(self, type_hint):
                    return self._sync_client

            adapter = AsyncBackendSyncAdapter(_SafeBackend())
            client = adapter.unwrap(SyncClient)
            ```
        """
    def exists(self, path: str) -> bool:
    def is_file(self, path: str) -> bool:
    def is_folder(self, path: str) -> bool:
    def read_bytes(self, path: str) -> bytes:
    def get_file_info(self, path: str) -> FileInfo:
    def get_folder_info(self, path: str) -> FolderInfo:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def check_health(self) -> None:
        """Submit a connectivity probe to the wrapped async backend.

        Not a no-op: the probe is forwarded to the wrapped
        ``AsyncBackend``, and any connectivity error
        it raises reaches the sync caller unchanged.

        Returns:
            ``None`` on success.

        Raises:
            BackendUnavailable: Or another backend-specific error if the
                wrapped backend's health check fails.
            RuntimeError: If the adapter is closed or called from a running
                event loop.

        Example:
            ```python
            try:
                adapter.check_health()
            except BackendUnavailable:
                ...  # handle connectivity failure
            ```
        """
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a forward-only chunk-pull stream.

        The returned stream pulls one async chunk per ``read()`` call; the
        file is never fully materialised in memory.  At most one
        ``__anext__`` is in-flight at a time.

        The stream is forward-only: ``seekable()`` returns ``False`` and no
        ``seek`` / ``tell`` / ``fileno`` methods are exposed.  Closing via
        the context manager or ``close()`` submits ``aclose()`` on the
        underlying async iterator so backend resources are released promptly.

        Args:
            path: Backend-relative key of the file to read.

        Returns:
            A forward-only ``io.RawIOBase`` stream over the file
            contents.

        Raises:
            RuntimeError: If the adapter is closed or called from a running
                event loop.
            NotFound: If *path* does not exist (propagated verbatim from the
                wrapped backend).

        Example:
            ```python
            with adapter.read("data/report.csv") as stream:
                header = stream.read(512)
            ```
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
    def glob(self, pattern: str) -> Iterator[FileInfo]:
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def open_atomic(self, path: str, *, overwrite: bool = False) -> AbstractContextManager[BinaryIO]:
    def close(self, timeout: float | None = 30.0) -> None:
        """Drain in-flight work, stop the loop, and join the daemon thread.

        Drain order: submit ``aclose`` on the wrapped backend → loop-drain
        in-flight tasks (repeating while new tasks appear, see below) → stop
        the private loop → join the daemon thread.

        The drain step repeats ``_drain_tasks`` until the private loop is
        quiet, **narrowing** (not eliminating) the window where a caller that
        passed ``_guard`` *before* the closed flag was set can still submit
        a coroutine.  Each pass snapshots outstanding tasks and waits for them;
        new tasks that arrive between snapshot and completion trigger another
        pass.  A residual TOCTOU gap remains: a thread that passes
        ``_guard`` after the final empty-snapshot check but before the loop
        is stopped will have its coroutine silently discarded when the loop
        stops — eliminating this gap would require serialising all submits
        against a shutdown lock.

        If *timeout* expires before the loop is drained, a single ``WARNING``
        record is emitted (message stem
        ``"AsyncBackendSyncAdapter close timed out"``) and the daemon thread
        is left for process-exit reaping.  Idempotent: subsequent calls are
        no-ops.

        Args:
            timeout: Maximum seconds to wait for in-flight work to finish and
                the background thread to join.  ``None`` means wait
                indefinitely.  Defaults to ``30.0``.
        """
    def __enter__(self) -> AsyncBackendSyncAdapter:
    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
```

[3/69] src/remote_store/_backend.py (526 rows, definitions)
---
```python
class Backend(abc.ABC):
    """Abstract base class for all storage backends.

    Every backend must implement all abstract methods. Backend-native
    exceptions must never leak — they must be mapped to ``remote_store`` errors.
    """
    # Subclasses must assign a CapabilitySet here; enforced by the conformance suite.
    CAPABILITIES: ClassVar[CapabilitySet]
    # BE-020 close posture (BK-298). ``False`` (the default) means the backend
    # is reusable after ``close()`` — lazy clients re-initialise on next use
    # (LocalBackend, MemoryBackend, SFTP, HTTP, SQL). ``True`` means ``close()``
    # is *terminal*: a subsequent operation raises ``BackendUnavailable`` rather
    # than silently re-opening resources (Azure, S3, Graph). The use-after-close
    # conformance lane gates on this flag.
    close_is_terminal: ClassVar[bool] = False
    @property
    @abc.abstractmethod
    def name(self) -> str:
        """Unique identifier for this backend type (e.g. ``'local'``, ``'s3'``)."""
    @property
    @abc.abstractmethod
    def capabilities(self) -> CapabilitySet:
        """Declared capabilities of this backend."""
    @abc.abstractmethod
    def exists(self, path: str) -> bool:
        """Check if a file or folder exists. Never raises ``NotFound``.

        Returns ``False`` if any ancestor of *path* is a file (file-as-directory-component),
        as traversal cannot proceed.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if a file or folder exists at *path*. ``False`` if the path
            does not exist or if any ancestor is a file instead of a directory.
        """
    @abc.abstractmethod
    def is_file(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing file.

        Returns ``False`` if the path does not exist, or if any ancestor of *path*
        is a file (file-as-directory-component).

        Args:
            path: Backend-relative key.

        Returns:
            ``True`` if *path* exists and is a file. ``False`` if the path does not
            exist or if any ancestor is a file instead of a directory.
        """
    @abc.abstractmethod
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing folder.

        Returns ``False`` if the path does not exist, or if any ancestor of *path*
        is a file (file-as-directory-component).

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if *path* exists and is a folder. ``False`` if the path does not
            exist or if any ancestor is a file instead of a directory.
        """
    @abc.abstractmethod
    def read(self, path: str) -> BinaryIO:
        """Open a file for reading and return a binary stream.

        Args:
            path: Backend-relative key.

        Returns:
            A readable binary stream.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* names a directory, not a file.
        """
    @abc.abstractmethod
    def read_bytes(self, path: str) -> bytes:
        """Read the full content of a file as bytes.

        Args:
            path: Backend-relative key.

        Returns:
            The file content.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* names a directory, not a file.
        """
    def read_seekable(self, path: str) -> BinaryIO:
        """Open a file for random-access reading and return a seekable stream.

        The default implementation delegates to ``read()``.  If the returned
        stream is already seekable, it is returned as-is.  Otherwise, the
        stream is spooled into a ``SpooledTemporaryFile`` (up to 8 MB in
        RAM, beyond that on disk) and returned positioned at byte 0.

        Backends MAY override to provide an optimized implementation.
        For example, ``AzureBackend`` returns a range reader that issues
        HTTP Range requests on each ``read()`` call.

        Args:
            path: Backend-relative key.

        Returns:
            A seekable binary stream positioned at byte 0.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* names a directory, not a file.
        """
    @abc.abstractmethod
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content to a file.

        Args:
            path: Backend-relative key.
            content: Data to write.
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-supplied key/value pairs to store alongside the
                file. Only honoured when the backend declares ``USER_METADATA``.

        Returns:
            A ``WriteResult`` with at least ``path`` and ``size`` populated.
            Backends declaring ``WRITE_RESULT_NATIVE`` also populate ``etag``,
            ``last_modified``, and where available ``digest`` and ``version_id``.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or if any slash-aligned
                ancestor of *path* exists as a regular file. Flat-namespace
                backends (S3, Azure non-HNS, SQL) cannot detect a file
                ancestor in O(1) and skip the check by default; the
                per-backend ``reject_write_under_file_ancestor`` opt-in
                enables it.
        """
    @abc.abstractmethod
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content atomically via temp file + rename.

        Args:
            path: Backend-relative key.
            content: Data to write.
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-supplied key/value pairs (see ``write()``).

        Returns:
            A ``WriteResult`` (same contract as ``write()``).

        Raises:
            CapabilityNotSupported: If backend lacks ``ATOMIC_WRITE``.
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or if any slash-aligned
                ancestor of *path* exists as a regular file (see ``write``).
        """
    @abc.abstractmethod
    def open_atomic(self, path: str, *, overwrite: bool = False) -> AbstractContextManager[BinaryIO]:
        """Yield a writable file object backed by a temporary location.

        On successful exit the temp file is atomically promoted to *path*.
        On exception the temp file is removed and *path* is untouched.

        Args:
            path: Backend-relative key.
            overwrite: If ``False``, raise if file already exists.

        Raises:
            AlreadyExists: If *path* exists and *overwrite* is ``False``.
            InvalidPath: If *path* names a directory, or if any slash-aligned
                ancestor of *path* exists as a regular file (see ``write``).
            CapabilityNotSupported: If the backend lacks ``ATOMIC_WRITE``.
        """
    @abc.abstractmethod
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a file.

        Args:
            path: Backend-relative key.
            missing_ok: If ``True``, do not raise when the file is absent.

        Raises:
            NotFound: If the file is missing and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a directory (type mismatch is
                not silenced by *missing_ok*).
        """
    @abc.abstractmethod
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Backend-relative key.
            recursive: If ``True``, delete all contents first.
            missing_ok: If ``True``, do not raise when absent.

        Raises:
            NotFound: If the folder is missing and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a file, not a folder.
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
        """
    @abc.abstractmethod
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """List files under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.
            recursive: If ``True``, include files in all subdirectories.
            max_depth: Optional maximum folder depth to traverse.  When set,
                backends that support native depth limiting prune traversal
                early.  Backends that ignore this parameter still produce
                correct results — the Store applies client-side filtering
                as a safety net.  ``None`` (default) defers to *recursive*.

        Returns:
            An iterator of ``FileInfo`` objects.
        """
    @abc.abstractmethod
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """List immediate subfolders under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An iterator of ``FolderEntry`` objects with ``.name`` and ``.path``.
        """
    @abc.abstractmethod
    def get_file_info(self, path: str) -> FileInfo:
        """Get metadata for a file.

        Args:
            path: Backend-relative key.

        Returns:
            A ``FileInfo`` with size, modification time, etc.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* names a directory, not a file.
        """
    @abc.abstractmethod
    def get_folder_info(self, path: str) -> FolderInfo:
        """Get metadata for a folder.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            A ``FolderInfo`` with file count, total size, etc.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If *path* names a file, not a folder.
        """
    @abc.abstractmethod
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename a file.

        When ``src == dst`` the call is a no-op (data preserved).

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            NotFound: If ``src`` does not exist.
            InvalidPath: If ``src`` or ``dst`` names a directory, or if any
                slash-aligned ancestor of ``dst`` exists as a regular file.
            AlreadyExists: If ``dst`` exists and ``overwrite`` is ``False``.
        """
    @abc.abstractmethod
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file.

        When ``src == dst`` the call is a no-op (data preserved).

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            NotFound: If ``src`` does not exist.
            InvalidPath: If ``src`` or ``dst`` names a directory, or if any
                slash-aligned ancestor of ``dst`` exists as a regular file.
            AlreadyExists: If ``dst`` exists and ``overwrite`` is ``False``.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield both files and folders under ``path`` in a single pass.

        Files are yielded as ``FileInfo`` objects, folders as
        ``FolderEntry`` objects. The default implementation chains
        ``list_files()`` and ``list_folders()``. Backends that can fetch
        both in a single I/O call should override this for efficiency.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An iterator of ``FileInfo`` (files) and ``FolderEntry`` (folders).
        """
    def glob(self, pattern: str) -> Iterator[FileInfo]:
        """Match files against a glob pattern.

        Non-abstract — backends with native glob support override this
        and add ``Capability.GLOB`` to their capability set.

        Args:
            pattern: Glob pattern (e.g., ``"data/*.csv"``, ``"**/*.txt"``).

        Raises:
            CapabilityNotSupported: If the backend lacks ``GLOB``.
        """
    def to_key(self, native_path: str) -> str:
        """Convert a backend-native path to a backend-relative key.

        Strips the backend's own root/prefix from the path. The default
        implementation is the identity function — backends with a native
        root (filesystem path, bucket prefix, base_path) override this.

        Args:
            native_path: Absolute or backend-native path string.

        Returns:
            Path relative to the backend's root.
        """
    def native_path(self, path: str) -> str:
        """Convert a backend-relative key to the backend-native path.

        The inverse of ``to_key()``. The default implementation is the
        identity function — backends with a native root (bucket, base_path)
        override this to prepend their prefix.

        Args:
            path: Backend-relative key.

        Returns:
            Backend-native path usable with the native handle from ``unwrap()``.
        """
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` describing how *path* maps to storage.

        Pure introspection -- no I/O is performed.  Backends override this
        to populate ``details`` with backend-specific context.

        Args:
            path: Backend-relative key.

        Returns:
            A frozen ``ResolutionPlan`` with ``kind``, ``backend``,
            ``key``, ``native_path``, and ``details``.
        """
    def check_health(self) -> None:  # noqa: B027
        """Verify the backend is reachable and credentials are valid.

        The default implementation is a no-op (always succeeds). Backends
        override this to perform a lightweight, non-destructive connectivity
        check using the cheapest possible read-only operation.

        Raises:
            PermissionDenied: If credentials are invalid.
            NotFound: If the bucket, container, or root path does not exist.
            BackendUnavailable: If the backend cannot be reached.
        """
    def close(self) -> None:  # noqa: B027
        """Release resources. Default is a no-op.

        Whether the backend may be reused afterwards is the
        ``close_is_terminal`` posture: the default backend is reusable;
        terminal backends raise ``BackendUnavailable`` on use-after-close.
        """
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the native backend handle if it matches the requested type.

        Args:
            type_hint: The expected type (e.g., ``fsspec.AbstractFileSystem``).

        Raises:
            CapabilityNotSupported: If backend cannot provide the requested type.
        """
```

[4/69] src/remote_store/_capabilities.py (182 rows, definitions)
---
```python
class Capability(enum.Enum):
    """Operations a backend may support.

    Most values gate one or more ``Store`` methods; some are quality
    flags that inform callers about backend behaviour without gating a
    specific method (see ``ATOMIC_MOVE``, ``SEEKABLE_READ``, ``LAZY_READ``).
    Use ``Store.supports()`` to query at runtime.

    Values:

    *Core I/O*

    - ``READ`` -- Stream or bulk-read file content.
      Gates ``Store.read()`` and ``Store.read_bytes()``.
    - ``WRITE`` -- Create or overwrite files.
      Gates ``Store.write()``.
    - ``DELETE`` -- Remove files and folders.
      Gates ``Store.delete()`` and ``Store.delete_folder()``.

    *Navigation*

    - ``LIST`` -- Enumerate files and subfolders.
      Gates ``Store.list_files()`` and ``Store.list_folders()``.
    - ``GLOB`` -- Native pattern matching against file paths.
      Gates ``Store.glob()``. Not all backends support this — use
      ``ext.glob.glob_files()`` as a portable fallback.

    *File operations*

    - ``MOVE`` -- Rename or relocate a file within the same backend.
      Gates ``Store.move()``.
    - ``COPY`` -- Duplicate a file within the same backend.
      Gates ``Store.copy()``.

    *Atomic variants*

    - ``ATOMIC_WRITE`` -- Write via temp-file-and-rename so readers never
      see partial content. Gates ``Store.write_atomic()`` and
      ``Store.open_atomic()``.
    - ``ATOMIC_MOVE`` -- Quality flag: ``move()`` is guaranteed atomic
      under concurrent access (e.g. Local via ``os.rename``, Memory under
      lock, SQL in a transaction). Does **not** gate a method — call
      ``store.supports(Capability.ATOMIC_MOVE)`` before relying on
      atomic rename semantics. Backends that implement move as
      copy-then-delete (e.g. S3, Azure non-HNS) do not declare this
      capability.

    *Metadata*

    - ``METADATA`` -- Retrieve file or folder metadata.
      Gates ``Store.get_file_info()`` and ``Store.get_folder_info()``.
    - ``USER_METADATA`` -- Store user-supplied key/value pairs alongside
      a file. Strict gate on the ``metadata=`` kwarg in
      ``Store.write()``, ``Store.write_text()``, and
      ``Store.write_atomic()``: passing a non-empty mapping to a backend
      that does not declare this capability raises
      ``CapabilityNotSupported`` before any I/O.
      ``metadata=None`` and ``metadata={}`` are always allowed.

    *Quality flags*

    - ``SEEKABLE_READ`` -- ``Store.read()`` always returns a seekable
      stream (``stream.seekable()`` is ``True``).  Backends that
      declare this capability return seekable streams from both
      ``read()`` and ``read_seekable()`` with zero overhead.  Absence of
      the flag means only that ``read()`` is forward-only -- not that
      seekable reads are unavailable.  On the sync ``Store``,
      ``read_seekable()`` is gated on ``READ`` alone, so it is served on
      every backend: a backend without the flag serves it either through a
      native override as cheap as the passthrough (the sync Azure backend
      issues one ranged download per ``read()``, with no temp-file spill)
      or through a ``SpooledTemporaryFile`` that copies the object first
      (HTTP, and any async backend bridged to sync).  The async API has no
      ``read_seekable()`` -- an async-native backend that omits the flag
      (``AsyncAzureBackend``, ``GraphBackend``) has no seekable read until
      bridged to sync.  The flag does not distinguish the native and
      spooled costs; the Azure backend guide's "Streaming and seekable
      reads" section documents the sync Azure range reader in full.
    - ``LAZY_READ`` -- ``read()`` fetches data lazily on demand from the
      native source rather than loading the entire file into memory before
      returning.  Backends that pre-load the full file contents (e.g.
      in-memory backends, SQL blob stores) do **not** declare this flag.
      Callers can use ``store.supports(Capability.LAZY_READ)`` to know
      whether partial reads avoid loading the entire file.
    - ``WRITE_RESULT_NATIVE`` -- Quality flag: the backend populates the
      rich fields of the returned ``WriteResult`` (``etag``,
      ``last_modified``, ``version_id``, and where applicable
      ``digest``) directly from its write response.  Does **not** gate
      any method — ``Store.write*()`` works on every backend.
      Backends without this flag return a ``WriteResult`` with only
      ``path`` and ``size`` populated (``source == "basic"``);
      ``metadata`` is governed independently by the ``USER_METADATA``
      capability and is not subject to this flag.
      Use ``store.supports(Capability.WRITE_RESULT_NATIVE)`` to decide
      whether to call ``store.get_file_info()`` after a write if you
      need the full metadata set.
    """
    # Core I/O
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    # Navigation
    LIST = "list"
    GLOB = "glob"
    # File operations
    MOVE = "move"
    COPY = "copy"
    # Atomic variants
    ATOMIC_WRITE = "atomic_write"
    ATOMIC_MOVE = "atomic_move"
    # Metadata
    METADATA = "metadata"
    USER_METADATA = "user_metadata"
    # Quality flags
    SEEKABLE_READ = "seekable_read"
    LAZY_READ = "lazy_read"
    WRITE_RESULT_NATIVE = "write_result_native"
class CapabilitySet:
    """Immutable set of capabilities declared by a backend.

    Args:
        capabilities: The set of supported capabilities.
    """
    __slots__ = ("_caps",)
    def __init__(self, capabilities: set[Capability]) -> None:
    def supports(self, cap: Capability) -> bool:
        """Check whether a capability is supported."""
    def require(self, cap: Capability, *, backend: str = "") -> None:
        """Raise if a capability is not supported.

        Raises:
            CapabilityNotSupported: If the capability is missing.
        """
    def __contains__(self, cap: object) -> bool:
    def __iter__(self) -> Iterator[Capability]:
    def __len__(self) -> int:
    def __repr__(self) -> str:
    def __setattr__(self, name: str, value: object) -> None:
    def __delattr__(self, name: str) -> None:
```

[5/69] src/remote_store/_config.py (426 rows, definitions)
---
```python
class Secret:
    """Immutable wrapper that prevents accidental exposure of credential strings.

    ``__repr__`` and ``__str__`` always return a masked value.
    Call ``.reveal()`` to obtain the plain-text credential.

    Args:
        value: The secret string to protect.

    Raises:
        TypeError: If *value* is not a ``str``.
    """
    __slots__ = ("_value",)
    def __init__(self, value: str) -> None:
    def reveal(self) -> str:
        """Return the plain-text secret."""
    def __repr__(self) -> str:
    def __str__(self) -> str:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    def __bool__(self) -> bool:
    def __setattr__(self, name: str, value: object) -> None:
    def __delattr__(self, name: str) -> None:
    def __reduce__(self) -> tuple[type[Secret], tuple[str]]:
class SecretRedactionFilter(logging.Filter):
    """Logging filter that replaces ``Secret`` instances in log record args with ``'***'``.

    Attach to any handler or logger to prevent credential leakage through
    %-style formatting:

    ```python
    handler.addFilter(SecretRedactionFilter())
    ```
    """
    def filter(self, record: logging.LogRecord) -> bool:
@dataclasses.dataclass(frozen=True)
class RetryPolicy:
    """Retry configuration for transient backend errors.

    Backends map these parameters to their native retry mechanisms.
    Backends that don't support a parameter silently ignore it.

    Args:
        max_attempts: Maximum number of attempts (including the initial).
            Set to 1 to disable retry.
        backoff_base: Base delay in seconds for exponential backoff.
        backoff_max: Maximum delay between retries in seconds.
        jitter: Maximum random jitter added to each delay in seconds.
            Set to 0.0 to disable jitter.
        timeout: Total wall-clock timeout in seconds for all attempts
            combined. ``None`` means no total timeout.
    """
    max_attempts: int = 3
    backoff_base: float = 1.0
    backoff_max: float = 60.0
    jitter: float = 1.0
    timeout: float | None = None
    def __post_init__(self) -> None:
    @classmethod
    def disabled(cls) -> RetryPolicy:
        """Return a policy that disables retry (single attempt, no backoff)."""
@dataclasses.dataclass(frozen=True)
class BackendConfig:
    """Describes a backend instance.

    Args:
        type: Backend type identifier (e.g. ``"local"``, ``"s3"``).
        options: Backend-specific configuration options.
        retry: Optional retry policy for transient errors.
    """
    type: str
    options: dict[str, object] = dataclasses.field(default_factory=dict)
    retry: RetryPolicy | None = None
@dataclasses.dataclass(frozen=True)
class StoreProfile:
    """Describes a named store.

    Args:
        backend: Name of the backend config to use.
        root_path: Path prefix for all operations.
        options: Store-specific options.
    """
    backend: str
    root_path: str = ""
    options: dict[str, object] = dataclasses.field(default_factory=dict)
@dataclasses.dataclass(frozen=True)
class RegistryConfig:
    """Top-level configuration container.

    Args:
        backends: Mapping of backend names to their configs.
        stores: Mapping of store names to their profiles.
    """
    backends: dict[str, BackendConfig] = dataclasses.field(default_factory=dict)
    stores: dict[str, StoreProfile] = dataclasses.field(default_factory=dict)
    def validate(self) -> None:
        """Validate that all store profiles reference existing backends.

        Raises:
            ValueError: If a store references a non-existent backend.
        """
    @classmethod
    def from_dict(cls, data: dict[str, object]) -> RegistryConfig:
        """Construct from a plain dict (e.g. parsed TOML/JSON).

        Args:
            data: Dict with ``backends`` and ``stores`` keys.
        """
    @classmethod
    def from_toml(
        cls,
        path: str | Path,
        *,
        table: tuple[str, ...] = (),
        resolve_env_vars: bool = False,
    ) -> RegistryConfig:
        """Load config from a TOML file.

        Args:
            path: Path to the TOML file.
            table: Dotted table path to extract config from.
                For ``pyproject.toml`` use ``table=("tool", "remote-store")``.
            resolve_env_vars: When ``True``, resolve ``${VAR}`` placeholders
                via ``resolve_env`` before constructing the config.

        Raises:
            ModuleNotFoundError: If ``tomllib`` is unavailable and
                ``tomli`` is not installed.
            KeyError: If a *table* key is not found, or if
                *resolve_env_vars* is ``True`` and a placeholder
                references an unset variable with no default.
        """
def resolve_env(
    data: dict[str, object],
    *,
    environ: Mapping[str, str] | None = None,
) -> dict[str, object]:
    """Resolve ``${VAR}`` placeholders in a config dict.

    Recursively walks *data* and replaces ``${VAR}`` with the value of
    the environment variable *VAR*. Use ``${VAR:-default}`` to provide a
    fallback when *VAR* is not set. Escape with ``$${`` to produce a
    literal ``${``.

    The original dict is never mutated; a deep copy with resolved values
    is returned.

    Args:
        data: Config dict (typically parsed from YAML/TOML).
        environ: Variable source. Defaults to ``os.environ``.

    Returns:
        A new dict with all placeholder strings resolved.

    Raises:
        KeyError: If a placeholder references a variable that is not set
            and has no default. The message includes the variable name
            and the config key path where it was found.
    """
```

[6/69] src/remote_store/_errors.py (177 rows, definitions)
---
```python
class RemoteStoreError(Exception):
    """Base class for all remote_store errors.

    Args:
        message: Human-readable error description.
        path: The path involved in the error, if any.
        backend: The backend name involved, if any.
    """
    def __init__(self, message: str = "", *, path: str | None = None, backend: str | None = None) -> None:
    def __str__(self) -> str:
    def __repr__(self) -> str:
class NotFound(RemoteStoreError):
    """Raised when a file or folder does not exist.

    Raised by read, delete, metadata, move, and copy operations when the
    target path does not exist.
    """
class AlreadyExists(RemoteStoreError):
    """Raised when a target already exists and overwrite is not allowed.

    Raised by write and copy operations when ``overwrite=False`` (the
    default) and the destination already exists.
    """
class PermissionDenied(RemoteStoreError):
    """Raised when access is denied by the storage backend.

    Raised by any Store or Backend method when the underlying storage
    system denies access (e.g., missing credentials, insufficient
    permissions on the bucket or container).
    """
class InvalidPath(RemoteStoreError):
    """Raised for malformed, unsafe, out-of-scope, or wrong-type paths.

    Raised by any method that validates paths: empty strings in file-targeted
    operations, paths containing ``..`` or null bytes, paths that fall
    outside the store's root scope, and paths that name the wrong type
    (e.g. a file operation on a directory path, or a folder operation on a
    file path).
    """
class CapabilityNotSupported(RemoteStoreError):
    """Raised when an operation requires an unsupported capability.

    Raised by capability-gated Store methods and by
    ``CapabilitySet.require()`` when a backend does not declare the
    needed capability. Check ``Store.supports()`` before calling
    capability-gated methods.

    Args:
        capability: The name of the unsupported capability.
    """
    def __init__(
        self,
        message: str = "",
        *,
        path: str | None = None,
        backend: str | None = None,
        capability: str = "",
    ) -> None:
    def __str__(self) -> str:
    def __repr__(self) -> str:
class DirectoryNotEmpty(RemoteStoreError):
    """Raised when a non-recursive delete targets a non-empty folder.

    Raised by ``Store.delete_folder()`` when ``recursive=False`` (the
    default) and the folder contains files or subfolders.
    """
class BackendUnavailable(RemoteStoreError):
    """Raised when the backend cannot be reached or initialized.

    Raised during backend construction or first operation when the
    storage service is unreachable (e.g., network error, invalid
    endpoint, missing container).
    """
class ResourceLocked(RemoteStoreError):
    """Raised when a target resource is locked by another session.

    The resource exists and the caller is authorised, but another session
    or process holds it (e.g. an open Office co-authoring session), so the
    operation cannot proceed now. Not retried by the default policy: the
    lock has no ``Retry-After`` and may last indefinitely. Maps from
    Microsoft Graph ``423 Locked``.
    """
```

[7/69] src/remote_store/_glob.py (108 rows, definitions)
---
```python
def extract_prefix(pattern: str) -> str:
    """Extract the longest non-wildcard directory prefix.

    >>> extract_prefix("data/2024/*.csv")
    'data/2024'
    >>> extract_prefix("**/*.csv")
    ''
    >>> extract_prefix("*.txt")
    ''
    """
def needs_recursive(pattern: str) -> bool:
    """Determine whether the pattern requires recursive listing.

    Returns ``True`` if the pattern contains ``**`` or if any non-final
    path segment contains wildcards.
    """
def pattern_to_regex(pattern: str) -> re.Pattern[str]:
    """Convert a glob pattern to a compiled regex.

    Supports:
    - ``*``    -> ``[^/]*`` (any chars except separator)
    - ``**/``  -> ``(?:.+/)?`` (zero or more path segments)
    - ``**``   (at end) -> ``.*`` (match everything)
    - ``?``    -> ``[^/]`` (single non-separator char)
    - ``[...]``  -> character class (passed through)
    - ``[!...]`` -> negated character class (``[!abc]`` -> ``[^abc]``)

    ``**`` must be a complete path segment (``**/``, ``/**``, or the
    entire pattern). Patterns like ``**error`` are not supported and
    raise ``ValueError``.
    """
```

[8/69] src/remote_store/_info.py (127 rows, definitions)
---
```python
class ExtensionInfo(TypedDict):
    """Information about a single extension."""
    available: bool
    extras: str | None
class InfoResult(TypedDict):
    """Structured result of the ``info`` function."""
    version: str
    backends: dict[str, BackendInfo]
    extensions: dict[str, ExtensionInfo]
def info() -> InfoResult:
    """Return a structured summary of available backends and extensions.

    Populates the backend registry, then probes each backend and optional
    extension for availability in the current environment.

    Returns:
        An ``InfoResult`` with keys ``version``, ``backends``, and
        ``extensions``.
    """
```

[9/69] src/remote_store/_models.py (205 rows, definitions)
---
```python
@typing.runtime_checkable
class PathEntry(typing.Protocol):
    """Shared interface for listing results -- every entry has a name and path."""
    @property
    def name(self) -> str:
        """Entry name (final path component)."""
    @property
    def path(self) -> RemotePath:
        """Normalized remote path."""
@dataclasses.dataclass(frozen=True)
class ContentDigest:
    """Verified content digest with known algorithm.

    Both ``algorithm`` and ``value`` are normalized to lowercase on
    construction.

    Attributes:
        algorithm: Hash algorithm name, always lowercase (e.g., ``"sha256"``).
        value: Lowercase hex-encoded digest, no prefix, no separators.
    """
    algorithm: str
    value: str
    def __post_init__(self) -> None:
@dataclasses.dataclass(frozen=True, eq=False)
class FileInfo:
    """Immutable snapshot of file metadata.

    Satisfies the ``PathEntry`` protocol.

    Attributes:
        path: Normalized remote path.
        name: File name (final path component).
        size: File size in bytes.
        modified_at: Last modification time.
        digest: Verified content digest with known algorithm.
        etag: Opaque backend-provided tag for change detection.
        content_type: Optional MIME type.
        metadata: User-supplied key/value metadata echoed from the backend.
        extra: Backend-specific metadata.
    """
    path: RemotePath
    name: str
    size: int
    modified_at: datetime
    digest: ContentDigest | None = None
    etag: str | None = None
    content_type: str | None = None
    metadata: Mapping[str, str] | None = None
    extra: dict[str, object] = dataclasses.field(default_factory=dict)
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
# Field-wise eq/hash by design (WR-001a) — intentionally different from the path-based eq=False on sibling models.
@dataclasses.dataclass(frozen=True)
class WriteResult:
    """Immutable snapshot of a completed write operation.

    Returned by ``Store.write()``, ``Store.write_text()``, and
    ``Store.write_atomic()``.

    Attributes:
        path: Normalized written path, store-relative.
        size: Bytes written.
        source: Provenance of the optional fields.
            ``"native"`` — the backend populated them from its write
            response; trust ``digest``, ``etag``, and ``last_modified``.
            ``"basic"`` — only ``path`` and ``size`` are reliable; call
            ``Store.head()`` or use ``ext.write`` helpers if you need
            more.
            ``"sidecar"`` — constructed by ``Store.head()`` from a
            subsequent ``get_file_info()`` call.
        digest: Content digest from the write — either a client-computed
            hash from ``ext.write`` helpers, or a hash echoed by the
            backend from its write response (e.g., Azure echoes the
            client-supplied MD5 as ``ContentDigest("md5", …)``).
            ``None`` when neither source applies.
        etag: Opaque backend change tag; semantics vary by backend.
        version_id: Immutable backend version identifier; ``None`` when
            the backend does not version objects.
        last_modified: Server timestamp from the write response; ``None``
            when the backend's write response omits it.
        metadata: Echo of user metadata stored with the object.
    """
    path: RemotePath
    size: int
    source: Literal["native", "basic", "sidecar"] = "basic"
    digest: ContentDigest | None = None
    etag: str | None = None
    version_id: str | None = None
    last_modified: datetime | None = None
    metadata: Mapping[str, str] | None = None
@dataclasses.dataclass(frozen=True, eq=False)
class FolderEntry:
    """Immutable folder identity returned by listing operations.

    Satisfies the ``PathEntry`` protocol.

    Attributes:
        path: Normalized remote path.
        name: Folder name (final path component).
    """
    path: RemotePath
    name: str
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
@dataclasses.dataclass(frozen=True, eq=False)
class FolderInfo:
    """Aggregated folder metadata.

    Satisfies the ``PathEntry`` protocol.

    Attributes:
        path: Normalized remote path.
        file_count: Number of files in the folder.
        total_size: Total size of all files in bytes.
        modified_at: Optional last modification time.
        extra: Backend-specific metadata.
    """
    path: RemotePath
    file_count: int
    total_size: int
    modified_at: datetime | None = None
    extra: dict[str, object] = dataclasses.field(default_factory=dict)
    @property
    def name(self) -> str:
        """Folder name (final path component).

        Unlike ``FileInfo`` and ``FolderEntry``, which store ``name``
        as a constructor field, this is a derived property (``self.path.name``)
        to avoid redundancy and keep ``name`` in sync with ``path``.
        """
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
```

[10/69] src/remote_store/_path.py (124 rows, definitions)
---
```python
class RemotePath:
    """An immutable, normalized path within a remote store.

    Args:
        raw: The raw path string to normalize and validate.

    Raises:
        InvalidPath: If the path is malformed or unsafe.
    """
    __slots__ = ("_path",)
    ROOT: ClassVar[RemotePath]
    def __init__(self, raw: str) -> None:
    @property
    def name(self) -> str:
        """Final component of the path."""
    @property
    def parent(self) -> RemotePath | None:
        """Parent path, or ``None`` if the path has only one component.

        Example: ``RemotePath("a/b").parent`` returns ``RemotePath("a")``,
        but ``RemotePath("a").parent`` returns ``None``.
        """
    @property
    def parts(self) -> tuple[str, ...]:
        """Tuple of path components."""
    @property
    def suffix(self) -> str:
        """File extension including the dot, or empty string."""
    def as_posix(self) -> str:
        """Return the path as a forward-slash string.

        Mirrors ``pathlib.PurePath.as_posix``. The path is always stored
        with forward slashes, so the result is identical to ``str(self)`` on
        every platform. ``RemotePath.ROOT.as_posix()`` returns ``"."``.
        """
    def __truediv__(self, other: str) -> RemotePath:
    def __str__(self) -> str:
    def __repr__(self) -> str:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    def __setattr__(self, name: str, value: object) -> None:
    def __delattr__(self, name: str) -> None:
    @classmethod
    def from_backend_path(cls, path: str) -> RemotePath:
        """Create a RemotePath, using ROOT for empty paths.

        Backends use this in ``get_folder_info`` to avoid duplicating
        the ``RemotePath(path) if path else RemotePath.ROOT`` pattern.
        """
```

[11/69] src/remote_store/_proxy.py (252 rows, definitions)
---
```python
class ProxyStore(Store):
    """Base class for Store proxies that delegate to an inner Store.

    All public ``Store`` methods delegate to ``self._inner`` by default.
    Subclasses override only the methods they intercept and must implement
    ``_wrap_child()`` to control how ``child()`` propagates wrapper behavior.

    ``ObservedStore`` and ``CachedStore`` are built on this base.
    Subclass it to build your own Store middleware.

    Args:
        inner: The Store instance to wrap.

    Example:
        ```python
        from remote_store import ProxyStore, Store

        class LoggingStore(ProxyStore):
            def read_bytes(self, path: str) -> bytes:
                print(f"Reading {path}")
                return self.inner.read_bytes(path)

            def _wrap_child(self, inner_child: Store) -> "LoggingStore":
                return LoggingStore(inner_child)
        ```
    """
    def __init__(self, inner: Store) -> None:
    @property
    def inner(self) -> Store:
        """The wrapped Store instance."""
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    def child(self, subpath: str) -> Store:
        """Return a child store wrapped with the same proxy behavior.

        Delegates to ``_wrap_child()`` which subclasses must implement.
        """
    def read(self, path: str) -> BinaryIO:
    def read_bytes(self, path: str) -> bytes:
    def read_seekable(self, path: str) -> BinaryIO:
    def read_text(self, path: str, *, encoding: str = "utf-8", errors: str = "strict") -> str:
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_text(
        self,
        path: str,
        text: str,
        *,
        encoding: str = "utf-8",
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
    def list_folders(
        self, path: str, *, pattern: str | None = None, max_depth: int | None = None
    ) -> Iterator[FolderEntry]:
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
    def glob(self, pattern: str) -> Iterator[FileInfo]:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def exists(self, path: str) -> bool:
    def is_file(self, path: str) -> bool:
    def is_folder(self, path: str) -> bool:
    def get_file_info(self, path: str) -> FileInfo:
    def get_folder_info(self, path: str, *, max_depth: int | None = None) -> FolderInfo:
    def head(self, path: str) -> WriteResult:
    def ping(self) -> None:
    def close(self) -> None:
    def unwrap(self, type_hint: type[T]) -> T:
    def native_path(self, key: str) -> str:
    def resolve(self, key: str) -> ResolutionPlan:
    def to_key(self, path: str) -> str:
    def supports(self, capability: Capability) -> bool:
```

[12/69] src/remote_store/_registry.py (191 rows, definitions)
---
```python
def register_backend(type_name: str, cls: type[Backend]) -> None:
    """Register a backend class for a given type string.

    Args:
        type_name: The type identifier (e.g. ``"local"``).
        cls: The backend class to instantiate.
    """
class Registry:
    """Manages backend lifecycle and provides access to named stores.

    Args:
        config: Optional configuration. Validates immediately.

    Raises:
        ValueError: If config is invalid.
    """
    def __init__(self, config: RegistryConfig | None = None) -> None:
    def __repr__(self) -> str:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    def get_store(self, name: str) -> Store:
        """Get a store by its profile name.

        Args:
            name: The store profile name.

        Raises:
            KeyError: If no store profile with this name exists.
        """
    def close(self) -> None:
        """Close all instantiated backends.

        If any backend raises during ``close()``, the remaining backends are
        still closed.  The first exception encountered is re-raised after all
        backends have been processed.
        """
    def __enter__(self) -> Registry:
    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
```

[13/69] src/remote_store/_resolution.py (59 rows, definitions)
---
```python
@dataclass(frozen=True)
class ResolutionPlan:
    """Describes how a key maps to its storage location.

    Returned by ``Backend.resolve()`` and ``Store.resolve()``. The plan
    captures the resolution strategy, backend identity, resolved key,
    native path, and backend-specific context -- all without performing
    any I/O.

    ``details`` is wrapped in ``types.MappingProxyType`` via
    ``__post_init__`` to prevent accidental mutation at runtime.

    Args:
        kind: Resolution strategy identifier (e.g. ``"local"``,
            ``"s3"``, ``"azure"``).
        backend: Human-readable backend identifier (typically
            ``Backend.name``).
        key: The resolved key (store-relative after
            ``Store.resolve()``, backend-relative after
            ``Backend.resolve()``).
        native_path: Backend-native location string (same as
            ``Backend.native_path()`` output).
        details: Backend-specific resolution context.  Immutable at
            runtime.  Values should be JSON-serializable primitives.
    """
    kind: str
    backend: str
    key: str
    native_path: str
    details: dict[str, Any]
    def __post_init__(self) -> None:
```

[14/69] src/remote_store/_retry.py (133 rows, definitions)
---
```python
def parse_retry_after(value: str | None, *, now: datetime | None = None) -> float | None:
    """Parse a ``Retry-After`` header into seconds, or ``None`` if unusable.

    Supports both forms RFC 7231 allows: delta-seconds (``"120"``) and an
    HTTP-date (``"Wed, 21 Oct 2025 07:28:00 GMT"``), the latter expressed as the
    remaining seconds until that instant (never negative). A date with no timezone
    token is assumed UTC rather than rejected. Returns ``None`` when the header is
    absent or unparseable, so the caller falls back to its computed backoff.

    Args:
        value: The raw header value, or ``None`` when absent.
        now: Reference instant for the HTTP-date delta. Defaults to the current
            UTC time; injectable for deterministic tests.
    """
def backoff_envelope(attempt: int, *, base: float, cap: float) -> float:
    """Return the deterministic exponential envelope for a 0-based *attempt*.

    ``min(base * 2**attempt, cap)`` — the un-jittered backoff for the wait that
    follows the attempt numbered *attempt* (``0`` is the first attempt).
    """
def equal_jitter_delay(attempt: int, *, base: float, cap: float, jitter: float, rng: Random | None = None) -> float:
    """Envelope plus an additive uniform ``[0, jitter]`` draw.

    The strategy used by ``graph_send`` and the sync HTTP retry loop: the
    deterministic envelope sets the floor, and the jitter spreads retriers above
    it. *rng* defaults to the module ``random``; inject a ``random.Random`` for
    deterministic tests.
    """
def full_jitter_delay(attempt: int, *, base: float, cap: float, rng: Random | None = None) -> float:
    """A uniform draw over the whole ``[0, envelope]`` band (full-jitter strategy).

    The strategy used by the Graph ``overwrite=True`` create-race re-attempt:
    randomising the *entire* delay desynchronises concurrent writers colliding in
    lockstep, where equal-jitter would still leave them clustered just above a
    shared floor. *rng* defaults to the module ``random``.
    """
def apply_retry_after(delay: float, retry_after: float | None) -> float:
    """Raise *delay* to at least *retry_after* (the server-requested floor).

    Returns *delay* unchanged when *retry_after* is ``None`` (no header, or one
    that did not parse).
    """
def budget_exhausted(*, elapsed: float, next_delay: float, timeout: float | None) -> bool:
    """Report whether the next wait would breach the wall-clock *timeout*.

    Look-ahead form: ``True`` when ``elapsed + next_delay >= timeout``. Pass
    ``next_delay=0.0`` for a plain "has the budget already run out?" check.
    ``timeout=None`` means no budget — always ``False``.
    """
```

[15/69] src/remote_store/_store.py (1106 rows, definitions)
---
```python
class Store:
    """A logical remote folder scoped to a root path.

    All path arguments are validated and prefixed with ``root_path``
    before being delegated to the backend.  Supports the context-manager
    protocol (``with Store(...) as s:``) which calls ``close()`` on exit.

    Args:
        backend: Backend instance (Local, S3, SFTP, Azure, Memory).
        root_path: Prefix prepended to every path.
            ``""`` means the backend root.
    """
    def __init__(self, backend: Backend, root_path: str = "") -> None:
    def read(self, path: str) -> BinaryIO:
        """Return a readable binary stream positioned at the start of *path*.

        The caller is responsible for closing the stream (or using a
        ``with`` block).

        Args:
            path: Store-relative file path.

        Returns:
            Readable binary stream positioned at byte 0.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
        """
    def read_bytes(self, path: str) -> bytes:
        """Read the entire file into memory and return ``bytes``.

        Args:
            path: Store-relative file path.

        Returns:
            The file content as ``bytes``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.

        Equivalent to ``read(path).read()``.
        """
    def read_seekable(self, path: str) -> BinaryIO:
        """Return a seekable binary stream for random-access reading.

        Always returns a seekable stream.  On backends that natively
        return seekable streams from ``read()``, this is zero-overhead.
        On backends whose ``read()`` is forward-only, the stream comes
        either from an optimized native implementation as cheap as the
        passthrough -- the sync Azure backend uses HTTP Range requests,
        one ranged download per ``read()``, with no temp-file spill -- or
        from a temporary-file spool that copies the object first (HTTP,
        and an async backend bridged to sync).

        Use ``read()`` for sequential streaming.  Use ``read_seekable()``
        when you need ``seek()`` / ``tell()`` -- for example, when
        passing the stream to PyArrow or other analytical readers.

        Args:
            path: Store-relative file path.

        Returns:
            A seekable binary stream positioned at byte 0.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
        """
    def read_text(self, path: str, *, encoding: str = "utf-8", errors: str = "strict") -> str:
        """Read the entire file and decode it as text.

        Args:
            path: Store-relative file path.
            encoding: Text encoding, any name accepted by ``codecs``.
            errors: Error handler: ``"strict"``, ``"ignore"``,
                ``"replace"``, ``"backslashreplace"``.  See
                ``codecs.register_error`` for custom handlers.

        Returns:
            The file content as ``str``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
            UnicodeDecodeError: If decoding fails with
                ``errors="strict"``.

        Equivalent to ``read_bytes(path).decode(encoding, errors)``.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write binary content to *path*.  Creates parent folders implicitly.

        Args:
            path: Store-relative file path.
            content: ``bytes`` or readable binary stream (``BinaryIO``).
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.
            metadata: Optional user-supplied key/value pairs to store
                alongside the file.  Requires ``Capability.USER_METADATA``.
                Keys must be non-empty ASCII strings with no leading
                underscore; total payload must not exceed 2048 bytes.

        Returns:
            ``WriteResult`` with at least ``path`` and ``size`` populated.

        Raises:
            AlreadyExists: If the file exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.
            ValueError: If *metadata* fails shape validation (keys and
                values must be ``str`` with bounded total size).
            CapabilityNotSupported: If *metadata* is non-empty and the
                backend lacks ``USER_METADATA``.
        """
    def write_text(
        self,
        path: str,
        text: str,
        *,
        encoding: str = "utf-8",
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write a string to *path*, encoded with the given encoding.

        Args:
            path: Store-relative file path.
            text: The string to write.
            encoding: Text encoding.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.
            metadata: Optional user-supplied key/value pairs (see ``write()``).

        Returns:
            ``WriteResult``.

        Raises:
            AlreadyExists: If the file exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.

        Equivalent to
        ``write(path, text.encode(encoding), overwrite=overwrite, metadata=metadata)``.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write binary content to *path* atomically.

        If the write fails or is interrupted, *path* is not left in a
        partial state.

        Args:
            path: Store-relative file path.
            content: ``bytes`` or readable binary stream (``BinaryIO``).
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.
            metadata: Optional user-supplied key/value pairs (see ``write()``).

        Returns:
            ``WriteResult``.

        Raises:
            CapabilityNotSupported: If backend lacks ``ATOMIC_WRITE``.
            AlreadyExists: If the file exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.
        """
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Context manager that yields a writable binary stream.

        The file is committed atomically on successful exit; on exception
        the partial write is discarded.

        Args:
            path: Store-relative file path.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.

        Returns:
            Writable binary stream.

        Raises:
            CapabilityNotSupported: If the backend lacks
                ``ATOMIC_WRITE``.
            AlreadyExists: If *path* exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.

        ```python
        with store.open_atomic("data/output.bin", overwrite=True) as f:
            f.write(b"chunk 1")
            f.write(b"chunk 2")
        # file is now visible at data/output.bin
        ```
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a single file.

        Args:
            path: Store-relative file path.
            missing_ok: If ``True``, silently succeeds when *path*
                does not exist.

        Raises:
            NotFound: If the file is missing and *missing_ok* is
                ``False``.
            InvalidPath: If *path* is empty, or if *path* names a directory
                (regardless of *missing_ok*).
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Store-relative folder path.  Must not be ``""``
                (root).
            recursive: If ``True``, delete all contents first.
                If ``False``, raises ``DirectoryNotEmpty`` when folder is
                non-empty.
            missing_ok: If ``True``, silently succeeds when *path*
                does not exist.

        Raises:
            NotFound: If the folder is missing and *missing_ok* is
                ``False``.
            DirectoryNotEmpty: If the folder is non-empty and
                *recursive* is ``False``.
            InvalidPath: If *path* is empty (cannot delete the store
                root), or if *path* names a file (use ``delete``
                instead).
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield ``FileInfo`` objects for files under *path*.

        Args:
            path: Store-relative folder path.
            recursive: Descend into subfolders.  Ignored when *max_depth*
                is set.
            pattern: Glob pattern to filter filenames
                (e.g. ``"*.csv"``).  Matched against each file's **name**
                (basename only).  For full path-based patterns, use
                ``ext.glob.glob_files()``.
            max_depth: Maximum folder depth to include.  ``0`` means files
                directly in *path* only; ``1`` adds files in its immediate
                subfolders, and so on.  ``None`` (default) defers to
                *recursive*.  When set, *recursive* is ignored.

        Returns:
            Iterator of ``FileInfo`` with store-relative paths.

        Raises:
            ValueError: If *max_depth* is negative.
        """
    def list_folders(
        self,
        path: str,
        *,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> Iterator[FolderEntry]:
        """Yield subfolders of *path* as ``FolderEntry`` objects.

        Args:
            path: Store-relative folder path.
            pattern: Glob pattern to filter folder names
                (e.g. ``"raw_*"``).  Matched against each folder's **name**
                (basename only) via ``fnmatch.fnmatch``.  Filters yielded
                results only — does **not** prune BFS traversal, so
                non-matching folders are still descended into.
            max_depth: Maximum folder depth to include.  ``None`` or ``0``
                returns immediate children only (default).  ``1`` adds
                grandchildren, and so on.  BFS traversal runs first;
                *pattern* filters what is yielded.

        Returns:
            Iterator of ``FolderEntry`` with ``.name`` and ``.path`` (store-relative).

        Raises:
            ValueError: If *max_depth* is negative.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield all immediate children (files and folders) of *path* in a single pass.

        Files are yielded as ``FileInfo``, folders as ``FolderEntry``.
        Both have ``.name`` and ``.path`` attributes (satisfying the
        ``PathEntry`` protocol) so callers can iterate uniformly.

        Args:
            path: Store-relative folder path.

        Returns:
            Iterator of ``FileInfo`` (files) and ``FolderEntry`` (folders).
        """
    def glob(self, pattern: str) -> Iterator[FileInfo]:
        """Yield files matching a glob *pattern*, using the backend's native glob implementation.

        Requires ``Capability.GLOB``.

        Args:
            pattern: Glob pattern (e.g. ``"data/**/*.parquet"``).

        Returns:
            Iterator of ``FileInfo`` with store-relative paths.

        Raises:
            CapabilityNotSupported: If the backend lacks ``GLOB``.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move (rename) a file from *src* to *dst*.

        File-only -- to move a folder, iterate its contents.

        Args:
            src: Source file path.
            dst: Destination file path.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *dst* exists.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists and *overwrite* is
                ``False``.
            InvalidPath: If *src* or *dst* is empty, or if *src* names a
                directory.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file from *src* to *dst*.

        File-only -- to copy a folder, iterate its contents.

        Args:
            src: Source file path.
            dst: Destination file path.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *dst* exists.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists and *overwrite* is
                ``False``.
            InvalidPath: If *src* or *dst* is empty, or if *src* names a
                directory.
        """
    def exists(self, path: str) -> bool:
        """Return ``True`` if *path* exists (file or folder).

        Never raises ``NotFound`` — returns ``False`` for missing paths instead.
        Also returns ``False`` if any ancestor of *path* is a file (file-as-directory-component),
        as traversal cannot proceed.

        Args:
            path: Store-relative path.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* exists and is a file.

        Returns ``False`` if any ancestor of *path* is a file (file-as-directory-component).

        Args:
            path: Store-relative path.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* exists and is a folder.

        Returns ``False`` if any ancestor of *path* is a file (file-as-directory-component).

        Args:
            path: Store-relative path.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return a ``FileInfo`` with size, modification time, and content type for a single file.

        Args:
            path: Store-relative file path.

        Returns:
            ``FileInfo``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
        """
    def get_folder_info(
        self,
        path: str,
        *,
        max_depth: int | None = None,
    ) -> FolderInfo:
        """Return a ``FolderInfo`` with aggregated size and file count for a folder.

        Args:
            path: Store-relative folder path.
            max_depth: Maximum folder depth to aggregate.  ``0`` means
                files directly in *path* only; ``1`` adds files in its
                immediate subfolders, and so on.  ``None`` (default)
                performs a full recursive traversal via the backend.

        Returns:
            ``FolderInfo``.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If *path* names a file (use ``get_file_info``
                instead).
            ValueError: If *max_depth* is negative.
        """
    def head(self, path: str) -> WriteResult:
        """Return a ``WriteResult`` snapshot of *path* via a metadata lookup.

        Gated on ``Capability.METADATA`` only — works on read-only backends
        that declare ``METADATA``.  The returned ``WriteResult`` has
        ``source="sidecar"`` (populated from a ``get_file_info()`` call, not
        from a write response).

        Args:
            path: Store-relative file path.

        Returns:
            ``WriteResult`` with ``source="sidecar"``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
            CapabilityNotSupported: If the backend lacks ``METADATA``.
        """
    def ping(self) -> None:
        """Verify that the backend is reachable.

        Raises:
            PermissionDenied: If credentials are invalid.
            NotFound: If the bucket, container, or root path does not
                exist.
            BackendUnavailable: If the backend cannot be reached.
        """
    def close(self) -> None:
        """Release backend resources.

        Called automatically when used as a context manager.
        """
    def child(self, subpath: str) -> Store:
        """Return a new ``Store`` scoped to *subpath* under the current root.

        The child shares the same backend instance.

        Args:
            subpath: Path segment to append to the current root.

        Returns:
            ``Store``.

        Raises:
            InvalidPath: If *subpath* is empty, contains ``..``
                segments, or includes null bytes.

        ```python
        data = store.child("data/2024")
        data.list_files("")  # lists files under <root>/data/2024/
        ```
        """
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the backend's native client object, cast to *type_hint*.

        Args:
            type_hint: The expected type of the native client
                (e.g. ``pyarrow.fs.FileSystem``).

        Returns:
            The native client.

        Raises:
            CapabilityNotSupported: If the backend cannot provide
                the requested type.

        ```python
        arrow_fs = store.unwrap(pyarrow.fs.FileSystem)
        ```
        """
    def resolve(self, key: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` describing how *key* maps to storage.

        Delegates to the backend's ``resolve()`` and rebases the key
        so that ``plan.key`` is the store-relative key, not the
        backend-relative path.

        Args:
            key: Store-relative path.  ``""`` resolves the store root.

        Returns:
            A frozen ``ResolutionPlan``.
        """
    def native_path(self, key: str) -> str:
        """Convert a store-relative *key* to the backend's native path representation.

        Inverse of ``to_key()``.

        Args:
            key: Store-relative path.

        Returns:
            Backend-native path (e.g. S3 object key, local filesystem path).
        """
    def to_key(self, path: str) -> str:
        """Convert a backend-native *path* to a store-relative key.

        Inverse of ``native_path()``.

        Args:
            path: Backend-native path string.

        Returns:
            Store-relative key.

        Raises:
            InvalidPath: If the path does not belong to this store.
        """
    def supports(self, capability: Capability) -> bool:
        """Check whether the backend supports a given ``Capability``.

        Args:
            capability: A ``Capability`` enum member.

        Returns:
            ``True`` if the backend declares this capability.

        ```python
        if store.supports(Capability.GLOB):
            results = store.glob("**/*.csv")
        ```
        """
    def __repr__(self) -> str:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    def __enter__(self) -> Store:
    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
```

[16/69] src/remote_store/_stream.py (135 rows, definitions)
[17/69] src/remote_store/_types.py (10 rows, definitions)
[18/69] src/remote_store/aio/__init__.py (29 rows, definitions)

[19/69] src/remote_store/aio/_async_backend.py (463 rows, definitions)
---
```python
class AsyncBackend(abc.ABC):
    """Abstract base class for all async storage backends.

    Every backend must implement all abstract methods. Backend-native
    exceptions must never leak -- they must be mapped to ``remote_store`` errors.
    """
    # Subclasses must assign a CapabilitySet here; enforced by the conformance suite.
    CAPABILITIES: ClassVar[CapabilitySet]
    # BE-020 close posture (BK-298). ``False`` (default) means the backend is
    # reusable after ``aclose()``; ``True`` means ``aclose()`` is terminal — a
    # subsequent operation raises ``BackendUnavailable`` (async Azure, Graph).
    # The use-after-close conformance lane gates on this flag.
    close_is_terminal: ClassVar[bool] = False
    async def __aenter__(self) -> AsyncBackend:
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
    @property
    @abc.abstractmethod
    def name(self) -> str:
        """Unique identifier for this backend type (e.g. ``'local'``, ``'s3'``)."""
    @property
    @abc.abstractmethod
    def capabilities(self) -> CapabilitySet:
        """Declared capabilities of this backend."""
    @abc.abstractmethod
    async def exists(self, path: str) -> bool:
        """Check if a file or folder exists. Never raises ``NotFound``.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if a file or folder exists at *path*.
        """
    @abc.abstractmethod
    async def is_file(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing file.

        Args:
            path: Backend-relative key.

        Returns:
            ``True`` if *path* exists and is a file.
        """
    @abc.abstractmethod
    async def is_folder(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing folder.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if *path* exists and is a folder.
        """
    @abc.abstractmethod
    async def read(self, path: str) -> AsyncIterator[bytes]:
        """Open a file for reading and return an async iterator of byte chunks.

        Args:
            path: Backend-relative key.

        Returns:
            An async iterator yielding byte chunks.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    @abc.abstractmethod
    async def read_bytes(self, path: str) -> bytes:
        """Read the full content of a file as bytes.

        Args:
            path: Backend-relative key.

        Returns:
            The file content.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    @abc.abstractmethod
    async def write(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content to a file.

        Args:
            path: Backend-relative key.
            content: Data to write.
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-defined string metadata.

        Returns:
            A ``WriteResult`` with size, path, and optional native fields.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory, or if any slash-aligned
                ancestor of ``path`` exists as a regular file. Flat-namespace
                backends (S3, Azure non-HNS, SQL) cannot detect a file
                ancestor in O(1) and skip the check by default; the
                per-backend ``reject_write_under_file_ancestor`` opt-in
                enables it.
        """
    @abc.abstractmethod
    async def write_atomic(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content atomically via temp file + rename.

        Args:
            path: Backend-relative key.
            content: Data to write.
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-defined string metadata.

        Returns:
            A ``WriteResult`` with size, path, and optional native fields.

        Raises:
            CapabilityNotSupported: If backend lacks ``ATOMIC_WRITE``.
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory, or if any slash-aligned
                ancestor of ``path`` exists as a regular file (see ``write``).
        """
    @abc.abstractmethod
    async def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a file.

        Args:
            path: Backend-relative key.
            missing_ok: If ``True``, do not raise when the file is absent.

        Raises:
            NotFound: If the file is missing and ``missing_ok`` is ``False``.
            InvalidPath: If the path is empty, or if ``path`` names a directory
                (regardless of ``missing_ok``).
        """
    @abc.abstractmethod
    async def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Backend-relative key.
            recursive: If ``True``, delete all contents first.
            missing_ok: If ``True``, do not raise when absent.

        Raises:
            InvalidPath: If ``path`` names an existing file (regardless of
                ``missing_ok`` — a type mismatch is not a missing file).
            NotFound: If the folder is missing and ``missing_ok`` is ``False``.
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
        """
    @abc.abstractmethod
    async def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> AsyncIterator[FileInfo]:
        """List files under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.
            recursive: If ``True``, include files in all subdirectories.
            max_depth: Optional maximum folder depth to traverse.  When set,
                backends that support native depth limiting prune traversal
                early.  Backends that ignore this parameter still produce
                correct results -- the Store applies client-side filtering
                as a safety net.  ``None`` (default) defers to *recursive*.

        Returns:
            An async iterator of ``FileInfo`` objects.
        """
    @abc.abstractmethod
    async def list_folders(self, path: str) -> AsyncIterator[FolderEntry]:
        """List immediate subfolders under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An async iterator of ``FolderEntry`` objects with ``.name`` and ``.path``.
        """
    @abc.abstractmethod
    async def get_file_info(self, path: str) -> FileInfo:
        """Get metadata for a file.

        Args:
            path: Backend-relative key.

        Returns:
            A ``FileInfo`` with size, modification time, etc.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    @abc.abstractmethod
    async def get_folder_info(self, path: str) -> FolderInfo:
        """Get metadata for a folder.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            A ``FolderInfo`` with file count, total size, etc.

        Raises:
            InvalidPath: If ``path`` names an existing file.
            NotFound: If the folder does not exist.
        """
    @abc.abstractmethod
    async def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename a file.

        ``src == dst`` is a no-op (the file is preserved unchanged).

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            InvalidPath: If ``src`` names a directory, ``dst`` names an
                existing directory, or any slash-aligned ancestor of ``dst``
                exists as a regular file.
            NotFound: If ``src`` does not exist.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and
                ``overwrite`` is ``False``.
        """
    @abc.abstractmethod
    async def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file.

        ``src == dst`` is a no-op (the file is preserved unchanged).

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            InvalidPath: If ``src`` names a directory, ``dst`` names an
                existing directory, or any slash-aligned ancestor of ``dst``
                exists as a regular file.
            NotFound: If ``src`` does not exist.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and
                ``overwrite`` is ``False``.
        """
    async def aclose(self) -> None:  # noqa: B027
        """Release resources. Default is a no-op."""
    async def check_health(self) -> None:  # noqa: B027
        """Verify the backend is reachable and credentials are valid.

        The default implementation is a no-op (always succeeds). Backends
        override this to perform a lightweight, non-destructive connectivity
        check using the cheapest possible read-only operation.

        Raises:
            PermissionDenied: If credentials are invalid.
            NotFound: If the bucket, container, or root path does not exist.
            BackendUnavailable: If the backend cannot be reached.
        """
    async def glob(self, pattern: str) -> AsyncIterator[FileInfo]:
        """Match files against a glob pattern.

        Non-abstract -- backends with native glob support override this
        and add ``Capability.GLOB`` to their capability set.

        Args:
            pattern: Glob pattern (e.g., ``"data/*.csv"``, ``"**/*.txt"``).

        Raises:
            CapabilityNotSupported: If the backend lacks ``GLOB``.
        """
    async def iter_children(self, path: str) -> AsyncIterator[FileInfo | FolderEntry]:
        """Yield both files and folders under ``path`` in a single pass.

        Files are yielded as ``FileInfo`` objects, folders as
        ``FolderEntry`` objects. The default implementation chains
        ``list_files()`` and ``list_folders()``. Backends that can fetch
        both in a single I/O call should override this for efficiency.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An async iterator of ``FileInfo`` (files) and ``FolderEntry`` (folders).
        """
    def to_key(self, native_path: str) -> str:
        """Convert a backend-native path to a backend-relative key.

        Strips the backend's own root/prefix from the path. The default
        implementation is the identity function -- backends with a native
        root (filesystem path, bucket prefix, base_path) override this.

        Args:
            native_path: Absolute or backend-native path string.

        Returns:
            Path relative to the backend's root.
        """
    def native_path(self, path: str) -> str:
        """Convert a backend-relative key to the backend-native path.

        The inverse of ``to_key()``. The default implementation is the
        identity function -- backends with a native root (bucket, base_path)
        override this to prepend their prefix.

        Args:
            path: Backend-relative key.

        Returns:
            Backend-native path usable with the native handle from ``unwrap()``.
        """
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` describing how *path* maps to storage.

        Pure introspection -- no I/O is performed.  Backends override this
        to populate ``details`` with backend-specific context.

        Args:
            path: Backend-relative key.

        Returns:
            A frozen ``ResolutionPlan`` with ``kind``, ``backend``,
            ``key``, ``native_path``, and ``details``.
        """
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the native backend handle if it matches the requested type.

        Args:
            type_hint: The expected type (e.g., ``fsspec.AbstractFileSystem``).

        Raises:
            CapabilityNotSupported: If backend cannot provide the requested type.
        """
```

[20/69] src/remote_store/aio/_async_store.py (1037 rows, definitions)
---
```python
class AsyncStore:
    """An async logical remote folder scoped to a root path.

    All path arguments are validated and prefixed with ``root_path``
    before being delegated to the backend.  Supports the async
    context-manager protocol (``async with AsyncStore(...) as s:``)
    which calls ``aclose()`` on exit.

    Args:
        backend: Async or sync backend instance.  Sync backends are
            auto-wrapped via ``SyncBackendAdapter``.
        root_path: Prefix prepended to every path.
            ``""`` means the backend root.
    """
    def __init__(self, backend: AsyncBackend | Backend, root_path: str = "") -> None:
    def read(self, path: str) -> AsyncIterator[bytes]:
        """Return an async iterator of byte chunks for *path*.

        The caller is responsible for consuming the iterator.
        Validation (capability check, path check) happens eagerly
        on call, not lazily on first iteration.

        Args:
            path: Store-relative file path.

        Returns:
            Async iterator of byte chunks.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
        """
    async def read_bytes(self, path: str) -> bytes:
        """Read the entire file into memory and return ``bytes``.

        Args:
            path: Store-relative file path.

        Returns:
            The file content as ``bytes``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.

        Equivalent to collecting all chunks from ``read(path)``.
        """
    async def read_text(self, path: str, *, encoding: str = "utf-8", errors: str = "strict") -> str:
        """Read the entire file and decode it as text.

        Args:
            path: Store-relative file path.
            encoding: Text encoding, any name accepted by ``codecs``.
            errors: Error handler: ``"strict"``, ``"ignore"``,
                ``"replace"``, ``"backslashreplace"``.  See
                ``codecs.register_error`` for custom handlers.

        Returns:
            The file content as ``str``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
            UnicodeDecodeError: If decoding fails with
                ``errors="strict"``.

        Equivalent to ``(await read_bytes(path)).decode(encoding, errors)``.
        """
    async def write(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write binary content to *path*.  Creates parent folders implicitly.

        Args:
            path: Store-relative file path.
            content: ``bytes`` or async iterator of ``bytes``.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.
            metadata: Optional user-defined string metadata.

        Returns:
            ``WriteResult`` with at least ``path`` and ``size`` populated.

        Raises:
            ValueError: If *metadata* contains invalid keys or values
                (see ``Store.write()`` for validation rules).
            CapabilityNotSupported: If *metadata* is non-empty and the
                backend lacks ``USER_METADATA``.
            AlreadyExists: If the file exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.
        """
    async def write_text(
        self,
        path: str,
        text: str,
        *,
        encoding: str = "utf-8",
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write a string to *path*, encoded with the given encoding.

        Args:
            path: Store-relative file path.
            text: The string to write.
            encoding: Text encoding.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.
            metadata: Optional user-defined string metadata.

        Raises:
            ValueError: If *metadata* contains invalid keys or values
                (see ``Store.write()`` for validation rules).
            CapabilityNotSupported: If *metadata* is non-empty and the
                backend lacks ``USER_METADATA``.
            AlreadyExists: If the file exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.

        Returns:
            ``WriteResult`` with at least ``path`` and ``size`` populated.

        Equivalent to
        ``await write(path, text.encode(encoding), overwrite=overwrite, metadata=metadata)``.
        """
    async def write_atomic(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write binary content to *path* atomically.

        If the write fails or is interrupted, *path* is not left in a
        partial state.

        Args:
            path: Store-relative file path.
            content: ``bytes`` or async iterator of ``bytes``.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *path* exists.
            metadata: Optional user-defined string metadata.

        Raises:
            ValueError: If *metadata* contains invalid keys or values
                (see ``Store.write()`` for validation rules).
            CapabilityNotSupported: If *metadata* is non-empty and the
                backend lacks ``USER_METADATA``, or if backend lacks
                ``ATOMIC_WRITE``.
            AlreadyExists: If the file exists and *overwrite* is
                ``False``.
            InvalidPath: If *path* is empty.

        Returns:
            ``WriteResult`` with at least ``path`` and ``size`` populated.
        """
    async def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a single file.

        Args:
            path: Store-relative file path.
            missing_ok: If ``True``, silently succeeds when *path*
                does not exist.

        Raises:
            NotFound: If the file is missing and *missing_ok* is
                ``False``.
            InvalidPath: If *path* is empty, or if *path* names a directory
                (regardless of *missing_ok*).
        """
    async def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Store-relative folder path.  Must not be ``""``
                (root).
            recursive: If ``True``, delete all contents first.
                If ``False``, raises ``DirectoryNotEmpty`` when folder is
                non-empty.
            missing_ok: If ``True``, silently succeeds when *path*
                does not exist.

        Raises:
            NotFound: If the folder is missing and *missing_ok* is
                ``False``.
            DirectoryNotEmpty: If the folder is non-empty and
                *recursive* is ``False``.
            InvalidPath: If *path* is empty (cannot delete the store
                root), or if *path* names a file (use ``delete``
                instead).
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> AsyncIterator[FileInfo]:
        """Yield ``FileInfo`` objects for files under *path*.

        Validation (capability check, max_depth) happens eagerly
        on call, not lazily on first iteration.

        Args:
            path: Store-relative folder path.
            recursive: Descend into subfolders.  Ignored when *max_depth*
                is set.
            pattern: Glob pattern to filter filenames
                (e.g. ``"*.csv"``).  Matched against each file's **name**
                (basename only).  For full path-based patterns, use
                ``ext.glob.glob_files()``.
            max_depth: Maximum folder depth to include.  ``0`` means files
                directly in *path* only; ``1`` adds files in its immediate
                subfolders, and so on.  ``None`` (default) defers to
                *recursive*.  When set, *recursive* is ignored.

        Returns:
            Async iterator of ``FileInfo`` with store-relative paths.

        Raises:
            ValueError: If *max_depth* is negative.
        """
    def list_folders(
        self,
        path: str,
        *,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> AsyncIterator[FolderEntry]:
        """Yield subfolders of *path* as ``FolderEntry`` objects.

        Validation (capability check, max_depth) happens eagerly
        on call, not lazily on first iteration.

        Args:
            path: Store-relative folder path.
            pattern: Glob pattern to filter folder names
                (e.g. ``"raw_*"``).  Matched against each folder's **name**
                (basename only) via ``fnmatch.fnmatch``.  Filters yielded
                results only — does **not** prune BFS traversal, so
                non-matching folders are still descended into.
            max_depth: Maximum folder depth to include.  ``None`` or ``0``
                returns immediate children only (default).  ``1`` adds
                grandchildren, and so on.  BFS traversal runs first;
                *pattern* filters what is yielded.

        Returns:
            Async iterator of ``FolderEntry`` with ``.name`` and ``.path`` (store-relative).

        Raises:
            ValueError: If *max_depth* is negative.
        """
    def iter_children(self, path: str) -> AsyncIterator[FileInfo | FolderEntry]:
        """Yield all immediate children (files and folders) of *path* in a single pass.

        Files are yielded as ``FileInfo``, folders as ``FolderEntry``.
        Both have ``.name`` and ``.path`` attributes (satisfying the
        ``PathEntry`` protocol) so callers can iterate uniformly.

        Validation (capability check) happens eagerly on call,
        not lazily on first iteration.

        Args:
            path: Store-relative folder path.

        Returns:
            Async iterator of ``FileInfo`` (files) and ``FolderEntry`` (folders).
        """
    def glob(self, pattern: str) -> AsyncIterator[FileInfo]:
        """Yield files matching a glob *pattern*, using the backend's native glob implementation.

        Requires ``Capability.GLOB``.  Validation (capability check) happens
        eagerly on call, not lazily on first iteration.

        Args:
            pattern: Glob pattern (e.g. ``"data/**/*.parquet"``).

        Returns:
            Async iterator of ``FileInfo`` with store-relative paths.

        Raises:
            CapabilityNotSupported: If the backend lacks ``GLOB``.
        """
    async def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move (rename) a file from *src* to *dst*.

        File-only -- to move a folder, iterate its contents.

        Args:
            src: Source file path.
            dst: Destination file path.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *dst* exists.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists and *overwrite* is
                ``False``.
            InvalidPath: If *src* or *dst* is empty, or if *src* names a
                directory.
        """
    async def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file from *src* to *dst*.

        File-only -- to copy a folder, iterate its contents.

        Args:
            src: Source file path.
            dst: Destination file path.
            overwrite: If ``False``, raises ``AlreadyExists`` when
                *dst* exists.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists and *overwrite* is
                ``False``.
            InvalidPath: If *src* or *dst* is empty, or if *src* names a
                directory.
        """
    async def exists(self, path: str) -> bool:
        """Return ``True`` if *path* exists (file or folder).

        Args:
            path: Store-relative path.
        """
    async def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* exists and is a file.

        Args:
            path: Store-relative path.
        """
    async def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* exists and is a folder.

        Args:
            path: Store-relative path.
        """
    async def get_file_info(self, path: str) -> FileInfo:
        """Return a ``FileInfo`` with size, modification time, and content type for a single file.

        Args:
            path: Store-relative file path.

        Returns:
            ``FileInfo``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
        """
    async def get_folder_info(
        self,
        path: str,
        *,
        max_depth: int | None = None,
    ) -> FolderInfo:
        """Return a ``FolderInfo`` with aggregated size and file count for a folder.

        Args:
            path: Store-relative folder path.
            max_depth: Maximum folder depth to aggregate.  ``0`` means
                files directly in *path* only; ``1`` adds files in its
                immediate subfolders, and so on.  ``None`` (default)
                performs a full recursive traversal via the backend.

        Returns:
            ``FolderInfo``.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If *path* names a file (use ``get_file_info``
                instead).
            ValueError: If *max_depth* is negative.
        """
    async def head(self, path: str) -> WriteResult:
        """Return a ``WriteResult`` snapshot of *path* via a metadata lookup.

        Async equivalent of ``Store.head()``.  Gated on ``Capability.METADATA``.
        Returns ``source="sidecar"``.

        Args:
            path: Store-relative file path.

        Returns:
            ``WriteResult`` with ``source="sidecar"``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* is empty, or if *path* names a directory.
            CapabilityNotSupported: If the backend lacks ``METADATA``.
        """
    async def ping(self) -> None:
        """Verify that the backend is reachable.

        Raises:
            PermissionDenied: If credentials are invalid.
            NotFound: If the bucket, container, or root path does not
                exist.
            BackendUnavailable: If the backend cannot be reached.
        """
    async def aclose(self) -> None:
        """Release backend resources.

        Called automatically when used as an async context manager.
        """
    def child(self, subpath: str) -> AsyncStore:
        """Return a new ``AsyncStore`` scoped to *subpath* under the current root.

        The child shares the same backend instance.

        Args:
            subpath: Path segment to append to the current root.

        Returns:
            ``AsyncStore``.

        Raises:
            InvalidPath: If *subpath* is empty, contains ``..``
                segments, or includes null bytes.

        ```python
        data = store.child("data/2024")
        async for fi in data.list_files(""):
            ...  # lists files under <root>/data/2024/
        ```
        """
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the backend's native client object, cast to *type_hint*.

        Args:
            type_hint: The expected type of the native client
                (e.g. ``pyarrow.fs.FileSystem``).

        Returns:
            The native client.

        Raises:
            CapabilityNotSupported: If the backend cannot provide
                the requested type.

        ```python
        arrow_fs = store.unwrap(pyarrow.fs.FileSystem)
        ```
        """
    def resolve(self, key: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` describing how *key* maps to storage.

        Delegates to the backend's ``resolve()`` and rebases the key
        so that ``plan.key`` is the store-relative key, not the
        backend-relative path.

        Args:
            key: Store-relative path.  ``""`` resolves the store root.

        Returns:
            A frozen ``ResolutionPlan``.
        """
    def native_path(self, key: str) -> str:
        """Convert a store-relative *key* to the backend's native path representation.

        Inverse of ``to_key()``.

        Args:
            key: Store-relative path.

        Returns:
            Backend-native path (e.g. S3 object key, local filesystem path).
        """
    def to_key(self, path: str) -> str:
        """Convert a backend-native *path* to a store-relative key.

        Inverse of ``native_path()``.

        Args:
            path: Backend-native path string.

        Returns:
            Store-relative key.

        Raises:
            InvalidPath: If the path does not belong to this store.
        """
    def supports(self, capability: Capability) -> bool:
        """Check whether the backend supports a given ``Capability``.

        Args:
            capability: A ``Capability`` enum member.

        Returns:
            ``True`` if the backend declares this capability.

        ```python
        if store.supports(Capability.GLOB):
            async for fi in store.glob("**/*.csv"):
                ...
        ```
        """
    def __repr__(self) -> str:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    async def __aenter__(self) -> AsyncStore:
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
```

[21/69] src/remote_store/aio/_sync_adapter.py (276 rows, definitions)
---
```python
class SyncBackendAdapter(AsyncBackend):
    """Wraps a synchronous ``Backend`` as an ``AsyncBackend``.

    Every blocking call is dispatched to the default executor via
    ``asyncio.to_thread``, keeping the event loop responsive.

    Args:
        backend: The synchronous backend instance to wrap.
    """
    # Universal upper bound — the wrapped backend's runtime capabilities() narrows this.
    CAPABILITIES: ClassVar[CapabilitySet] = CapabilitySet(set(Capability))
    def __init__(self, backend: _SyncBackend) -> None:
    @property
    def name(self) -> str:
        """Backend identifier, forwarded from the wrapped backend."""
    @property
    def capabilities(self) -> CapabilitySet:
        """Capability set, forwarded from the wrapped backend."""
    def to_key(self, native_path: str) -> str:
        """Convert a native path to a backend-relative key."""
    def native_path(self, path: str) -> str:
        """Convert a backend-relative key to the native path."""
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a resolution plan for *path*."""
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the native backend handle."""
    async def exists(self, path: str) -> bool:
        """Check if a file or folder exists."""
    async def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing file."""
    async def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing folder."""
    async def read_bytes(self, path: str) -> bytes:
        """Read the full content of a file as bytes.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    async def get_file_info(self, path: str) -> FileInfo:
        """Get metadata for a file.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    async def get_folder_info(self, path: str) -> FolderInfo:
        """Get metadata for a folder.

        Raises:
            InvalidPath: If ``path`` names an existing file.
            NotFound: If the folder does not exist.
        """
    async def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename a file.

        Raises:
            InvalidPath: If ``src`` names a directory or ``dst`` names an
                existing directory.
            NotFound: If ``src`` does not exist.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and
                ``overwrite`` is ``False``.
        """
    async def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file.

        Raises:
            InvalidPath: If ``src`` names a directory or ``dst`` names an
                existing directory.
            NotFound: If ``src`` does not exist.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and
                ``overwrite`` is ``False``.
        """
    async def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a file.

        Raises:
            NotFound: If the file is missing and ``missing_ok`` is ``False``.
            InvalidPath: If the path is empty, or if ``path`` names a directory
                (regardless of ``missing_ok``).
        """
    async def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Raises:
            InvalidPath: If ``path`` names an existing file (regardless of
                ``missing_ok`` — a type mismatch is not a missing file).
            NotFound: If the folder is missing and ``missing_ok`` is ``False``.
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
        """
    async def check_health(self) -> None:
        """Verify the backend is reachable."""
    async def read(self, path: str) -> AsyncIterator[bytes]:
        """Open a file for reading and yield chunks asynchronously.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    async def write(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content to a file.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory.
        """
    async def write_atomic(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content atomically via temp file + rename.

        Raises:
            CapabilityNotSupported: If backend lacks ``ATOMIC_WRITE``.
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory.
        """
    async def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> AsyncIterator[FileInfo]:
        """List files under *path*."""
    async def list_folders(self, path: str) -> AsyncIterator[FolderEntry]:
        """List immediate subfolders under *path*."""
    async def glob(self, pattern: str) -> AsyncIterator[FileInfo]:
        """Match files against a glob pattern."""
    async def iter_children(self, path: str) -> AsyncIterator[FileInfo | FolderEntry]:
        """Yield both files and folders under *path*."""
    async def aclose(self) -> None:
        """Release resources held by the wrapped backend."""
```

[22/69] src/remote_store/aio/_types.py (7 rows, definitions)
[23/69] src/remote_store/aio/backends/__init__.py (19 rows, definitions)

[24/69] src/remote_store/aio/backends/_azure.py (1400 rows, definitions)
---
```python
class AsyncAzureBackend(AsyncBackend):
    """Async Azure Storage backend.

    Uses the async Blob SDK for non-HNS accounts (plain Blob Storage, Azurite)
    and the async DataLake SDK for HNS accounts (ADLS Gen2) to get atomic
    rename and real directory support. Whether the account is HNS is declared
    explicitly via the required ``hns`` argument -- the backend does not probe
    for it.

    Args:
        container: Azure Storage container name (required, non-empty).
        hns: Whether the storage account has Hierarchical Namespace
            enabled (ADLS Gen2). **Required** -- there is no default and
            no runtime auto-detection. Pass ``True`` for ADLS Gen2
            accounts (atomic rename, real directories) or ``False`` for
            flat Blob Storage. Use ``AzureUtils.adetect_hns()`` to discover
            the value once if you do not already know it.
        account_name: Storage account name.
        account_url: Full account URL (e.g. ``https://myaccount.dfs.core.windows.net``).
        account_key: Storage account key.
        sas_token: Shared Access Signature token.
        connection_string: Azure Storage connection string.
        credential: Any credential object (e.g. ``DefaultAzureCredential()``).
        client_options: Additional options passed to service clients.
            The library sets ``max_single_put_size``, ``max_block_size``,
            and ``min_large_block_upload_threshold`` defaults for streaming
            memory discipline; user-supplied values take precedence.
        retry: Retry policy for transient failures.
        max_concurrency: Maximum number of parallel connections for
            uploads and downloads (default ``1`` -- sequential).
        reject_write_under_file_ancestor: If ``True``, ``write`` /
            ``write_atomic`` / ``move`` / ``copy`` HEAD each
            slash-aligned ancestor of the target path on non-HNS
            accounts and raise ``InvalidPath`` on the first regular-file
            hit, matching the cross-backend contract that hierarchical
            filesystems enforce natively. On HNS accounts the kwarg
            short-circuits: ``hdi_isfolder`` rejects the operation
            natively, and the backend detects the file ancestor on that
            rejection and re-raises it as ``InvalidPath``, so HNS delivers
            the cross-backend contract with or without the kwarg set.
            Default ``False``.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    __mirror__: ClassVar[type[AzureBackend]] = AzureBackend
    # M1 (BK-298): aclose() is terminal — a use-after-close raises
    # BackendUnavailable rather than silently re-initialising. Gates the
    # use-after-close conformance lane.
    close_is_terminal: ClassVar[bool] = True
    def __init__(
        self,
        container: str,
        *,
        hns: bool | None = None,
        account_name: str | None = None,
        account_url: str | None = None,
        account_key: str | Secret | None = None,
        sas_token: str | Secret | None = None,
        connection_string: str | Secret | None = None,
        credential: Any | None = None,
        client_options: dict[str, Any] | None = None,
        retry: RetryPolicy | None = None,
        max_concurrency: int = 1,
        reject_write_under_file_ancestor: bool = False,
    ) -> None:
    @property
    def name(self) -> str:
        """Unique identifier for this backend type."""
    @property
    def capabilities(self) -> CapabilitySet:
        """Declared capabilities of this backend."""
    async def check_health(self) -> None:
        """Verify the backend is reachable and credentials are valid.

        Raises:
            PermissionDenied: If credentials are invalid.
            NotFound: If the container does not exist.
            BackendUnavailable: If the backend cannot be reached.
        """
    def to_key(self, native_path: str) -> str:
        """Convert a backend-native path to a backend-relative key.

        Args:
            native_path: Absolute or backend-native path string.

        Returns:
            Path relative to the backend's root.
        """
    def native_path(self, path: str) -> str:
        """Convert a backend-relative key to the backend-native path.

        Args:
            path: Backend-relative key.

        Returns:
            Backend-native path (``container/path``).
        """
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with Azure-specific details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="async-azure"`` and ``details`` containing
            ``container`` and ``account_url``.
        """
    async def exists(self, path: str) -> bool:
        """Check if a file or folder exists.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if a file or folder exists at *path*.
        """
    async def is_file(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing file.

        Args:
            path: Backend-relative key.

        Returns:
            ``True`` if *path* exists and is a file.
        """
    async def is_folder(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing folder.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if *path* exists and is a folder.
        """
    async def read(self, path: str) -> AsyncIterator[bytes]:
        """Open a file for reading and return an async iterator of byte chunks.

        Args:
            path: Backend-relative key.

        Returns:
            An async iterator yielding byte chunks.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If ``path`` names a directory (HNS accounts only).
        """
    async def read_bytes(self, path: str) -> bytes:
        """Read the full content of a file as bytes.

        Args:
            path: Backend-relative key.

        Returns:
            The file content.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If ``path`` names a directory (HNS accounts only).
        """
    async def write(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content to a file.

        Args:
            path: Backend-relative key.
            content: Data to write (bytes or async iterator of bytes).
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-defined string metadata.

        Returns:
            ``WriteResult`` with native Azure fields (``etag``, ``last_modified``,
            etc.) populated from the SDK upload response.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory.
        """
    async def write_atomic(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content atomically via temp file + rename.

        For non-HNS accounts, direct upload is atomic (PUT semantics).
        For HNS accounts, write to temp file via DFS then atomic rename.

        Args:
            path: Backend-relative key.
            content: Data to write.
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-defined string metadata.

        Returns:
            ``WriteResult`` with native Azure fields populated from the SDK
            response (non-HNS) or from ``get_file_properties()`` after rename
            (HNS).

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory.
        """
    async def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a file.

        Args:
            path: Backend-relative key.
            missing_ok: If ``True``, do not raise when the file is absent.

        Raises:
            NotFound: If the file is missing and ``missing_ok`` is ``False``.
            InvalidPath: If ``path`` names a directory (HNS accounts only).
        """
    async def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Backend-relative key.
            recursive: If ``True``, delete all contents first.
            missing_ok: If ``True``, do not raise when absent.

        Raises:
            NotFound: If the folder is missing and ``missing_ok`` is ``False``.
            InvalidPath: If ``path`` names a file (use ``delete`` instead).
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
        """
    async def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> AsyncIterator[FileInfo]:
        """List files under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.
            recursive: If ``True``, include files in all subdirectories.
            max_depth: Optional maximum folder depth to traverse.

        Returns:
            An async iterator of ``FileInfo`` objects.
        """
    async def list_folders(self, path: str) -> AsyncIterator[FolderEntry]:
        """List immediate subfolders under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An async iterator of ``FolderEntry`` objects.
        """
    async def iter_children(self, path: str) -> AsyncIterator[FileInfo | FolderEntry]:
        """Yield both files and folders under ``path`` in a single pass.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An async iterator of ``FileInfo`` (files) and ``FolderEntry`` (folders).
        """
    async def glob(self, pattern: str) -> AsyncIterator[FileInfo]:
        """Match files against a glob pattern.

        Args:
            pattern: Glob pattern (e.g., ``"data/*.csv"``, ``"**/*.txt"``).

        Returns:
            An async iterator of matching ``FileInfo`` objects.
        """
    async def get_file_info(self, path: str) -> FileInfo:
        """Get metadata for a file.

        Args:
            path: Backend-relative key.

        Returns:
            A ``FileInfo`` with size, modification time, etc.

        Raises:
            InvalidPath: If ``path`` names a directory (HNS: ``hdi_isfolder=true``).
            NotFound: If the file does not exist.
        """
    async def get_folder_info(self, path: str) -> FolderInfo:
        """Get metadata for a folder.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            A ``FolderInfo`` with file count, total size, etc.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If ``path`` names a file (use ``get_file_info`` instead).
        """
    async def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename a file.

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            NotFound: If ``src`` does not exist.
            InvalidPath: If ``src`` or ``dst`` names a directory (HNS only).
            AlreadyExists: If ``dst`` exists and ``overwrite`` is ``False``.
        """
    async def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file.

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            NotFound: If ``src`` does not exist.
            InvalidPath: If ``src`` or ``dst`` names a directory (HNS only).
            AlreadyExists: If ``dst`` exists and ``overwrite`` is ``False``.
        """
    async def aclose(self) -> None:
        """Release all Azure SDK client resources."""
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the native async FileSystemClient if it matches the requested type.

        Args:
            type_hint: The expected type.

        Returns:
            The native async client instance matching *type_hint*.

        Raises:
            CapabilityNotSupported: If backend cannot provide the requested type.
        """
    def __del__(self) -> None:
    def __repr__(self) -> str:
```

[25/69] src/remote_store/aio/backends/_graph/__init__.py (16 rows, definitions)

[26/69] src/remote_store/aio/backends/_graph/auth.py (333 rows, definitions)
---
```python
class GraphAuth:
    """MSAL-backed token provider for ``GraphBackend``.

    Selects the OAuth flow from the supplied credentials: a ``client_secret``
    or ``client_certificate`` selects client-credentials (app-only); their
    absence selects device-code (interactive). The resulting bearer token is
    reachable synchronously through ``get_token()`` (and through calling the
    instance directly) and asynchronously through ``aget_token()``, which
    offloads the blocking MSAL work off the event loop and single-flights
    concurrent acquisitions — prefer it on the event loop.

    Args:
        tenant_id: Entra tenant id, or ``"consumers"`` / ``"common"`` /
            ``"organizations"`` for the device-code multi-tenant authorities.
        client_id: Application (client) id of the Entra app registration.
        client_secret: Client secret for client-credentials. Accepts a
            ``Secret`` and is masked in ``repr``.
        client_certificate: Certificate dict for client-credentials, as
            MSAL's ``client_credential`` mapping. Mutually exclusive with
            ``client_secret``.
        scopes: Override the default scope set. Client-credentials defaults
            to ``["https://graph.microsoft.com/.default"]``; device-code
            defaults to the delegated ``["Files.ReadWrite", "User.Read"]``
            (the signed-in user's OneDrive — consumer-compatible). Add
            ``Sites.ReadWrite.All`` for delegated SharePoint access.
        cache_path: Override the MSAL token-cache file location. Defaults to
            ``<user_config_dir("remote-store")>/graph_token_cache.json``. The
            cache is persisted multi-process-safely: a sibling
            ``<cache_path>.lockfile`` coordinates concurrent writers so a
            shared default cache survives the common multi-worker deployment.
        prompt_callback: Invoked with the MSAL device-flow dict on device-code
            login; defaults to printing ``flow["message"]``.

    Raises:
        ValueError: If ``tenant_id`` or ``client_id`` is empty, or if both
            ``client_secret`` and ``client_certificate`` are supplied.
    """
    def __init__(
        self,
        tenant_id: str,
        client_id: str,
        *,
        client_secret: str | Secret | None = None,
        client_certificate: dict[str, Any] | None = None,
        scopes: Sequence[str] | None = None,
        cache_path: str | None = None,
        prompt_callback: Callable[[dict[str, Any]], None] | None = None,
    ) -> None:
    @property
    def authority(self) -> str:
        """The MSAL authority URL derived from ``tenant_id``."""
    def get_token(self) -> str:
        """Acquire (or silently refresh) a bearer token.

        The token-provider callable the backend invokes. Re-invoking it after
        a ``401`` refreshes through MSAL's cache. The acquisition writes the
        refreshed cache through to disk itself — the backing
        ``PersistedTokenCache`` persists under a cross-process lock on every
        change, so no explicit flush is needed here.

        Raises:
            PermissionDenied: If MSAL returns no token (auth failure); the
                ``error_description`` is included, never the secret. A typed
                ``RemoteStoreError`` so a failure surfacing mid-``read`` /
                ``write`` stays catchable via ``except RemoteStoreError``.
        """
    def __call__(self) -> str:
        """Return a bearer token — makes the instance a token-provider callable."""
    async def aget_token(self) -> str:
        """Acquire a bearer token without blocking the event loop.

        The async token-provider entry point: pass ``token_provider=auth.aget_token``
        (a bound async method is a ``Callable[[], Awaitable[str]]``) so the backend
        awaits it instead of calling the synchronous instance. It offloads the
        synchronous MSAL acquisition — including any contended token-cache lock
        wait — to a worker thread so sibling coroutines keep running, and
        single-flights concurrent callers: N coroutines acquiring at once share
        **one** acquisition rather than each hitting the identity provider (which
        also dedupes the one-shot refresh after a ``401``).

        Use this on the event loop; reuse the synchronous ``get_token`` /
        ``__call__`` from synchronous wiring code that has no running loop. The
        single-flight state is bound to the loop the first caller runs on — share
        one instance per loop, mirroring the backend's own single-loop posture.
        Cancelling one caller (e.g. a ``gather`` sibling that times out) neither
        cancels the shared acquisition nor disturbs the other joiners.

        Raises:
            PermissionDenied: Propagated from ``get_token`` to the owning
                caller and every joiner sharing the in-flight acquisition; a
                later call retries afresh.
        """
    def flush_cache(self) -> None:
        """Best-effort no-op retained for the ``GraphBackend.close()`` hook.

        The token cache is a ``PersistedTokenCache`` that writes through to
        disk under a cross-process lock on every acquisition, so there is
        nothing to flush at close. The method stays because
        ``GraphBackend.close()`` invokes it duck-typed and user-supplied
        providers may implement their own. Never raises — teardown must not
        fail.
        """
    def __repr__(self) -> str:
```

[27/69] src/remote_store/aio/backends/_graph/backend.py (1375 rows, definitions)
---
```python
class GraphBackend(AsyncBackend):
    """Async Microsoft Graph backend over OneDrive / SharePoint / Teams files.

    A single instance targets one drive, identified by an immutable
    ``drive_id``. Items are addressed by ``/``-rooted POSIX path; transport is
    ``httpx``; auth is a token-provider callable (the built-in ``GraphAuth``
    helper, or any user-supplied callable).

    Args:
        drive_id: Opaque Graph drive id. Resolve one from a URL / "me" /
            Teams channel with ``GraphUtils.resolve_drive_id``.
        token_provider: ``Callable[[], str]`` or ``Callable[[], Awaitable[str]]``
            returning a bearer token, invoked lazily (never in ``__init__``).
        base_url: Graph API root (default ``https://graph.microsoft.com/v1.0``).
        http_client: Reuse an existing ``httpx.AsyncClient``; the caller owns
            its lifecycle and ``close()`` does not close it. When omitted, one
            is created lazily on first use and closed by ``close()``.
        retry: Retry policy for transient failures; ``None`` uses the default
            ``RetryPolicy()`` profile.
        upload_chunk_size: Upload-session chunk size; must be a positive
            multiple of 320 KiB and strictly less than 60 MiB (Graph's
            per-request ceiling). Default 10 MiB.
        copy_timeout: Wall-clock budget for copy/move monitor polling, or
            ``None`` for no backend-imposed ceiling. When set, must be a
            positive float.
        base_path: Optional drive subfolder to scope every operation under.
            When set, all keys are addressed relative to this folder and keys
            returned by listing / ``to_key`` stay relative to it, so the backend
            behaves as if ``base_path`` were its root. Defaults to the drive root.
        client_options: Extra options passed through to the internal
            ``httpx.AsyncClient``. (When a future revision adds an explicit
            httpx-level constructor parameter, it takes precedence over a
            ``client_options`` key of the same name; the backend has no such
            parameter today, so this is passthrough only.)

    Raises:
        ValueError: For an empty ``drive_id``, a non-callable
            ``token_provider``, a non-string ``base_path``, an
            ``upload_chunk_size`` that is not a positive 320 KiB multiple
            below 60 MiB, or a non-positive ``copy_timeout``.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _GRAPH_CAPABILITIES
    # M1 (BK-298): aclose() is terminal (BUG-219) — a use-after-close raises
    # BackendUnavailable. Declares the posture the conformance lane gates on.
    close_is_terminal: ClassVar[bool] = True
    def __init__(
        self,
        drive_id: str,
        *,
        token_provider: TokenProvider,
        base_url: str = _DEFAULT_BASE_URL,
        http_client: httpx.AsyncClient | None = None,
        retry: RetryPolicy | None = None,
        upload_chunk_size: int = _DEFAULT_UPLOAD_CHUNK_SIZE,
        copy_timeout: float | None = None,
        base_path: str = "",
        client_options: dict[str, Any] | None = None,
    ) -> None:
    @property
    def name(self) -> str:
        """Unique identifier for this backend type — ``"graph"``."""
    @property
    def capabilities(self) -> CapabilitySet:
        """Declared capabilities of this backend."""
    @property
    def drive_id(self) -> str:
        """The immutable target drive id."""
    def native_path(self, path: str) -> str:
        """Return the Graph item-by-path metadata endpoint for ``path``.

        ``/drives/{drive_id}/root:{encoded_path}:`` with each segment
        percent-encoded. The empty key returns ``/drives/{drive_id}/root:``
        (Graph's drive-root form, no trailing colon delimiter). A configured
        ``base_path`` is prepended so every key is scoped under it.
        """
    def to_key(self, native_path: str) -> str:
        """Strip the drive-root prefix/delimiter and decode back to a key.

        Inverse of ``native_path``: removes ``/drives/{drive_id}/root:`` and the
        trailing ``:`` delimiter, then percent-decodes each segment. Inputs
        without the prefix are returned unchanged; the drive root maps to ``""``.
        """
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` carrying the drive id and base URL."""
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the underlying ``httpx.AsyncClient`` (native-handle escape hatch).

        Raises:
            CapabilityNotSupported: For any type other than
                ``httpx.AsyncClient``.
        """
    async def aclose(self) -> None:
        """Close the owned HTTP client, cancel pollers, and flush the auth cache.

        Safe to call multiple times. A caller-supplied ``http_client`` is left
        open — the caller owns it. Any in-flight copy/move monitor poller is
        cancelled cooperatively, and any upload session a ``write()`` left
        mid-chunk-loop is aborted via best-effort ``DELETE`` — every cleanup
        error is swallowed so ``close()`` never raises. The server-side copy /
        move continues (Graph monitor URLs have no cancel endpoint).

        After this returns, any new op raises ``BackendUnavailable`` via the
        ``_client`` guard, and any op still in flight surfaces a typed error
        rather than a bare ``RuntimeError`` from the closing client.
        """
    async def read(self, path: str) -> AsyncIterator[bytes]:
        """Stream file content from the pre-signed download URL.

        Fetches item metadata first (so the directory check happens before any
        byte is yielded), then streams the response body from
        ``@microsoft.graph.downloadUrl`` with no ``Range`` header (the URL is
        pre-signed, so no ``Authorization`` header is attached either). If the
        read is interrupted — the URL expires, or the connection drops mid-body —
        the stream re-fetches metadata for a fresh URL and resumes from the next
        unread byte with a ``Range`` request, provided the ``eTag`` is unchanged.

        Raises:
            NotFound: If the path does not exist.
            InvalidPath: If the path names a folder.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: If the download URL is missing, the file changed
                mid-read (``eTag`` mismatch), the pre-signed host returns a
                non-success status, or the download fails at the transport level
                (connect/read timeout, DNS, reset).
        """
    async def read_bytes(self, path: str) -> bytes:
        """Read full file content as bytes.

        Delegates to ``read`` so the directory check and not-found mapping live
        in one place.

        Raises:
            NotFound: If the path does not exist.
            InvalidPath: If the path names a folder.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: As for ``read`` (missing URL, expiry, eTag change,
                host or transport failure).
        """
    async def exists(self, path: str) -> bool:
        """Return ``True`` if a file or folder exists at *path*.

        Any 404 — including a drive-identity ``resourceNotFound`` — is
        suppressed to ``False``; never raises ``NotFound``. A misconfigured
        or deleted drive therefore probes as missing rather than raising;
        it surfaces on the first error-raising operation.

        Raises:
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* exists and carries the ``file`` facet.

        A missing item returns ``False`` (any 404 is suppressed, including a
        drive-identity ``resourceNotFound``).

        Raises:
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* exists and carries the ``folder`` facet.

        A missing item returns ``False`` (any 404 is suppressed, including a
        drive-identity ``resourceNotFound``). The drive root (``""``) carries
        the ``folder`` facet and reports ``True``.

        Raises:
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def get_file_info(self, path: str) -> FileInfo:
        """Return file metadata mapped from the Graph ``driveItem`` body.

        ``file.hashes`` rides ``FileInfo.extra["graph.file.hashes"]``;
        ``metadata`` is ``None`` (user metadata is not declared) and ``digest``
        is left unset.

        Raises:
            NotFound: If the path does not exist.
            InvalidPath: If the path names a folder.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def iter_children(self, path: str) -> AsyncIterator[FileInfo | FolderEntry]:
        """Yield immediate files and folders under *path* in a single pass.

        Overrides the default (which would chain ``list_files`` + ``list_folders``
        and double the round-trips) to consume one ``/children`` response:
        ``folder``-faceted items become ``FolderEntry``, ``file``-faceted items
        become ``FileInfo``, in the order Graph returns them. A missing or file
        path yields nothing.

        Raises:
            PermissionDenied: If the token is rejected or lacks access
                (401/403), surfaced during iteration.
            BackendUnavailable: On throttling, 5xx, or transport failure,
                surfaced during iteration.
        """
    async def list_files(
        self, path: str, *, recursive: bool = False, max_depth: int | None = None
    ) -> AsyncIterator[FileInfo]:
        """List files under *path*.

        When ``max_depth`` is set it governs traversal depth and ``recursive`` is
        ignored; otherwise ``recursive=True`` walks the subtree unbounded and the
        default lists only immediate files. A missing or file path yields nothing.

        Raises:
            PermissionDenied: If the token is rejected or lacks access
                (401/403), surfaced during iteration.
            BackendUnavailable: On throttling, 5xx, or transport failure,
                surfaced during iteration.
        """
    async def list_folders(self, path: str) -> AsyncIterator[FolderEntry]:
        """List immediate subfolders under *path*.

        A missing or file path yields nothing.

        Raises:
            PermissionDenied: If the token is rejected or lacks access
                (401/403), surfaced during iteration.
            BackendUnavailable: On throttling, 5xx, or transport failure,
                surfaced during iteration.
        """
    async def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregated folder metadata: recursive file count + total size.

        Validates the path first — a missing item raises ``NotFound`` and a file
        item raises ``InvalidPath``. ``file_count`` and ``total_size`` aggregate
        every file in the subtree; ``modified_at`` is the folder item's own
        timestamp.

        Raises:
            NotFound: If the path does not exist.
            InvalidPath: If the path names a file.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def write(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write a file, creating intermediate folders implicitly.

        Content ``<= 4 MiB`` uses ``PUT /content``; larger content uploads via a
        chunked session. An ``AsyncIterator`` of unknown length is spooled to
        size it first (the session ``Content-Range`` needs a known total). The
        returned ``WriteResult`` is ``source="native"``, populated from the
        ``driveItem`` Graph returns. ``overwrite`` maps to Graph's
        ``@microsoft.graph.conflictBehavior`` (``replace`` vs ``fail``).

        On ``overwrite=True`` (``conflictBehavior=replace``) a concurrent create
        of the *same not-yet-existing* key can still draw a ``409
        nameAlreadyExists`` (live-reproduced on consumer OneDrive): the loser's
        replace lands while the winner's create is in flight. The write
        re-attempts the replace a bounded number of times — the winner's create
        has committed by the time the 409 returns, so the re-issue overwrites it.
        On the large-file upload-session path the same race can surface
        *mid-session* instead: the session opens cleanly, then a racing replace
        swaps the item and the next chunk ``PUT`` draws a ``404``; under
        ``overwrite=True`` that is treated as the same create-race signal and the
        write re-opens a fresh session and retries. Large-file retries re-upload the
        whole body, so this convergence is best-effort: under sustained concurrent
        same-key overwrite a loser may still exhaust the budget and raise
        ``AlreadyExists`` (last-writer-wins still holds — one writer's content lands
        intact). A terminal conflict (a SharePoint-backed drive that rejects the
        replace outright) likewise keeps 409-ing and surfaces ``AlreadyExists`` once
        the budget is spent; the retry re-issues the replace, it never swallows the
        conflict.

        Raises:
            AlreadyExists: If the file exists and ``overwrite=False``, or an
                ``overwrite=True`` replace keeps drawing a ``409`` past the bounded
                re-attempt budget (a SharePoint-backed replace-rejection, or a loser
                in a sustained concurrent large-file same-key overwrite).
            InvalidPath: If the path names the drive root, an existing folder,
                or descends through a file ancestor.
            CapabilityNotSupported: If a non-empty ``metadata=`` reaches the
                backend directly (``USER_METADATA`` is not declared).
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On 5xx / throttling / transport failure, or a
                Graph contract gap (missing ``uploadUrl`` / ``nextExpectedRanges``).
            ResourceLocked: If the item is locked mid-session. The session
                stays valid server-side, but its credentialed URL is not
                exposed in the exception.
        """
    async def write_atomic(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Atomic write — delegates to ``write``.

        Graph's own write paths already provide the no-partial-content guarantee
        (``PUT /content`` is service-atomic; an upload session commits only on the
        final chunk), so no client-side temp-rename is taken. The ``WriteResult``
        shape, the ``metadata=`` gate, and the raised exceptions are inherited
        verbatim from ``write``.
        """
    async def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a file (Graph moves it to the recycle bin).

        Fetches the item first so a folder is rejected before any ``DELETE`` is
        issued (a bare ``DELETE`` on a folder would remove the folder and its
        contents). A missing item is ``NotFound`` unless ``missing_ok``.

        Raises:
            NotFound: If the file does not exist and ``missing_ok`` is ``False``.
            InvalidPath: If the path names a folder (use ``delete_folder``).
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder, optionally requiring it to be empty.

        ``recursive=True`` is a single ``DELETE`` — Graph removes the folder and
        all contents atomically server-side. ``recursive=False`` first checks the
        folder is empty (via the item's ``folder.childCount``, falling back to a
        ``/children`` probe when the count is absent) and raises
        ``DirectoryNotEmpty`` if not.

        Raises:
            NotFound: If the folder does not exist and ``missing_ok`` is ``False``.
            InvalidPath: If the path names a file.
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On throttling, 5xx, or transport failure.
        """
    async def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file, awaiting the async monitor to completion.

        Graph answers ``POST copy`` with ``202 Accepted`` and a ``Location``
        monitor URL; this polls it to a terminal state (bounded by
        ``copy_timeout``). ``src == dst`` short-circuits after a single
        existence-confirming ``GET``. A destination conflict surfaces as
        ``AlreadyExists`` (or ``InvalidPath`` for a folder / file-ancestor target)
        when ``overwrite`` is ``False``.

        Raises:
            NotFound: If ``src`` does not exist.
            InvalidPath: If ``src`` names a folder, or ``dst`` names an existing
                folder or descends through a file ancestor.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On a ``202`` without a ``Location`` monitor URL, a
                ``copy_timeout`` expiry, or a transient/5xx failure.
        """
    async def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename a file, awaiting the monitor when Graph goes async.

        Graph answers ``PATCH driveItem`` synchronously in most cases (``200``);
        a large-item move may return ``202`` with a monitor URL (the async trigger
        is item size / server-side replication, not crossing folders), which is
        polled to completion exactly as ``copy`` does. ``src == dst``
        short-circuits after one ``GET``. Item identity (id / eTag / property
        bag) is preserved by Graph; the backend issues no compensating writes.

        Raises:
            NotFound: If ``src`` does not exist.
            InvalidPath: If ``src`` names a folder, or ``dst`` names an existing
                folder or descends through a file ancestor.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If the token is rejected or lacks access to the
                item (401/403).
            BackendUnavailable: On a ``202`` without a ``Location`` monitor URL, a
                ``copy_timeout`` expiry, or a transient/5xx failure.
        """
    def __repr__(self) -> str:
```

[28/69] src/remote_store/aio/backends/_graph/http.py (491 rows, definitions)
---
```python
def mask_headers(headers: Mapping[str, str]) -> dict[str, str]:
    """Return a copy of *headers* with sensitive values replaced by ``"***"``.

    The ``Authorization`` bearer token must never reach a log record or an
    exception message. Any backend code that logs request/response headers
    routes them through this helper first.
    """
def redact_presigned_url(url: str) -> str:
    """Return *url* with its query (and fragment) stripped — for pre-signed URLs.

    Graph's pre-signed URLs carry their own credential in the query string, so a
    token must never reach an exception message or log record. Wherever such a URL
    would otherwise surface for diagnosis it is routed through this helper,
    dropping the query while the scheme / host / path still identify the
    operation. Three call sites: the request DEBUG log (the upload-session chunk
    PUTs and the abort DELETE run unauthenticated, their credential in the URL),
    the copy/move monitor timeout message, and the upload-session ``423`` handler.
    The read download URL is the same credential class but is never surfaced in a
    message or log (it is streamed directly, not sent via ``graph_send``), so it
    needs no redaction.
    """
def error_code(body: object) -> str | None:
    """Extract the Graph ``error.code`` string from a parsed JSON body.

    Returns ``None`` when the body is not a Graph error envelope (no
    ``{"error": {"code": ...}}`` shape), which the caller treats as an
    unclassifiable response.
    """
def classify_graph_error(
    status: int,
    code: str | None,
    *,
    path: str = "",
    backend: str = BACKEND_NAME,
    scope: Literal["item", "drive", "probe"] = "item",
) -> RemoteStoreError:
    """Map an HTTP status plus Graph ``error.code`` to a ``remote_store`` error.

    The single mapping table for the backend. ``scope`` disambiguates the
    ``404`` case: an item/path-scoped ``itemNotFound`` is a per-item
    ``NotFound``, while a drive-scoped ``404`` (or ``resourceNotFound``,
    Graph's drive-identity code, at any URL scope) is a backend-identity
    failure mapped to ``BackendUnavailable``. ``"probe"`` is the type-probe
    scope (``exists`` / ``is_file`` / ``is_folder``): every ``404`` maps to
    ``NotFound`` regardless of ``code``, so the drive-identity escalation
    cannot escape a probe that suppresses ``NotFound`` to ``False``.

    Never inspects the error *message* — only ``status`` and ``code``.
    """
def classify_graph_error_code(
    code: str | None,
    *,
    path: str = "",
    backend: str = BACKEND_NAME,
) -> RemoteStoreError:
    """Map a status-less Graph ``error.code`` to a ``remote_store`` error.

    A failed async operation (copy / may-be-async move) reports its failure as an
    ``error.code`` string in the monitor poll body with no HTTP status of its own.
    This reuses the single ``classify_graph_error`` table via the known
    code→status reverse map; an unknown or missing code maps to
    ``BackendUnavailable`` (the operation failed and the cause is not classifiable).
    """
def response_json(response: httpx.Response) -> object | None:
    """Return the parsed JSON body, or ``None`` for a non-JSON / empty body.

    Used by the write path, which inspects ``409`` conflict bodies the flat
    status→error table is deliberately blind to.
    """
def discriminate_write_conflict(body: object, path: str, *, backend: str = BACKEND_NAME) -> RemoteStoreError:
    """Map a ``409`` write-conflict body to a typed error.

    Graph's ``409 nameAlreadyExists`` alone does not discriminate the
    precondition outcomes, so the structured body (facets on the conflicting
    item, never the error *message*) is inspected:

    * a ``folder`` facet on the **target** name → the target is an existing
      folder → ``InvalidPath``;
    * a ``file`` facet on an **ancestor** name (file-as-directory-component) →
      ``InvalidPath`` regardless of ``overwrite``;
    * otherwise the target is a file at the requested name → ``AlreadyExists``.

    The ancestor and folder cases are overwrite-independent. On the normal
    ``overwrite=True`` path ``conflictBehavior=replace`` overwrites an existing
    file (``200``), so a plain target-file does not reach this discriminator —
    only the structurally-invalid cases do. The exception is a known
    SharePoint-backed-drive divergence: some such drives return ``409`` for a
    plain file even under ``replace``, and it then falls through here to
    ``AlreadyExists`` despite the overwrite request (a documented hard backend
    limitation, not guarded — see the backend write-path spec).
    """
async def acquire_token(token_provider: TokenProvider) -> str:
    """Invoke *token_provider*, awaiting it when it is an async callable.

    Supports both ``Callable[[], str]`` and ``Callable[[], Awaitable[str]]``.
    The callable is invoked lazily by the caller — never from
    ``GraphBackend.__init__``.
    """
async def graph_send(
    client: httpx.AsyncClient,
    method: str,
    url: str,
    *,
    token_provider: TokenProvider,
    path: str = "",
    scope: Literal["item", "drive", "probe"] = "item",
    retry: RetryPolicy | None = None,
    return_on: frozenset[int] = frozenset(),
    authenticated: bool = True,
    **kwargs: Any,
) -> httpx.Response:
    """Send an authenticated Graph request and map failures to typed errors.

    Attaches ``Authorization: Bearer <token>`` from *token_provider*, sends the
    request, and on a non-2xx response raises the mapped ``remote_store`` error.
    A ``401 InvalidAuthenticationToken`` triggers one token refresh and a single
    re-issue; a second ``401`` raises ``PermissionDenied``. Transport-level
    failures map to ``BackendUnavailable``.

    Retry: when *retry* is supplied with ``max_attempts > 1``, a transient
    response (a retryable ``5xx`` / ``429``, or a transport error) is retried
    with exponential backoff ``min(backoff_max, backoff_base * 2**attempt)`` plus
    uniform ``[0, jitter]``; a ``Retry-After`` header (HTTP-date or delta-seconds)
    raises the wait to at least its value. ``retry.timeout`` bounds the whole
    loop. Terminal mappings (``PermissionDenied`` / ``NotFound`` /
    ``AlreadyExists`` / ``ResourceLocked`` / insufficient-storage) raise on the
    first attempt — they do not clear on short-term retry. ``retry=None`` is a
    single attempt (the default for every call site until it opts in), so the
    auth refresh is the only re-issue.

    ``return_on`` lists non-2xx statuses the caller wants to classify itself:
    a response whose status is in the set is **returned raw** instead of
    mapped-and-raised, and is never treated as transient (so it is not retried).
    The write path uses this to inspect a ``409`` body for the
    folder / ancestor-file / already-exists discrimination and to surface a
    mid-session ``423`` with the unfinished session context — cases the flat
    status→error table cannot disambiguate. The default empty set leaves every
    existing call site byte-identical.

    ``authenticated=False`` sends no bearer token (and skips the ``401`` refresh)
    for a pre-signed, self-authenticating target such as the upload-session
    ``uploadUrl`` — which lives on a different host and rejects a cross-host
    ``Authorization`` header, exactly like the ``downloadUrl`` reads.

    Caveat: both the auth refresh and the retry loop re-issue the request with
    the same ``kwargs``, so they are only safe when the body is replayable. A
    streaming body (an ``AsyncIterator`` ``content=`` / generator ``data=``) is
    consumed by the first attempt and would replay empty — the write path must
    re-materialise such a body, and must not pass a retry policy that would
    re-send it.
    """
async def iter_pages(
    client: httpx.AsyncClient,
    url: str,
    *,
    token_provider: TokenProvider,
    path: str = "",
    retry: RetryPolicy | None = None,
) -> AsyncIterator[dict[str, Any]]:
    """Yield each page body of a Graph collection, following ``@odata.nextLink``.

    Terminates when a page omits ``@odata.nextLink``. An empty ``value`` array
    carrying a ``nextLink`` is followed, not treated as the end. A ``nextLink``
    that is malformed, or that points to a different scheme/host than the
    original request, is a Graph contract violation and maps to
    ``BackendUnavailable`` rather than being followed: each page is re-fetched
    through ``graph_send`` (which attaches the bearer token), so following a
    cross-host link would leak the token to an unrelated host.

    Each page fetch is retried per *retry* (threaded straight into
    ``graph_send``); ``None`` keeps the single-attempt default.
    """
```

[29/69] src/remote_store/aio/backends/_graph/items.py (129 rows, definitions)
---
```python
def is_folder_item(item: Mapping[str, Any]) -> bool:
    """Return ``True`` when *item* carries the Graph ``folder`` facet."""
def is_file_item(item: Mapping[str, Any]) -> bool:
    """Return ``True`` when *item* carries the Graph ``file`` facet."""
def download_url(item: Mapping[str, Any]) -> str | None:
    """Return the pre-signed ``@microsoft.graph.downloadUrl``, or ``None``."""
def parse_graph_datetime(value: object) -> datetime:
    """Parse a Graph RFC 3339 timestamp into a timezone-aware ``datetime``.

    ``datetime.fromisoformat`` accepts the trailing ``Z`` only on Python 3.11+;
    the ``>=3.10`` floor needs it normalised to ``+00:00`` first. A missing or
    unparseable value falls back to the UTC epoch so ``modified_at`` stays a
    real ``datetime`` (Graph reliably returns the field, so this is defensive).
    """
def item_to_fileinfo(item: Mapping[str, Any], key: str) -> FileInfo:
    """Map a file ``driveItem`` to a ``FileInfo`` for store key *key*.

    ``file.hashes`` rides ``extra["graph.file.hashes"]`` when present; ``digest``
    is left unset (no canonical-hash selection in v1) and ``metadata`` is
    ``None`` (the backend does not declare user metadata).
    """
def item_to_write_result(
    item: Mapping[str, Any],
    key: str,
    size: int,
    metadata: Mapping[str, str] | None,
) -> WriteResult:
    """Map a write-response ``driveItem`` to a native ``WriteResult`` for *key*.

    Both ``PUT /content`` and the final upload-session chunk return a full
    ``driveItem`` body; this populates ``source="native"`` with ``size``,
    ``etag`` (cleaned to match ``get_file_info``), and ``last_modified``.
    ``version_id`` rides the SharePoint ``listItem`` version where Graph
    surfaces one, ``None`` otherwise. ``digest`` is left ``None`` — no
    canonical hash is selected from ``file.hashes`` in v1. ``metadata`` echoes
    the caller's input mapping (``None`` when none was supplied).

    *size* is the byte count the backend wrote (authoritative even when the
    response omits or under-reports ``size``), not re-derived from the body.
    """
```

[30/69] src/remote_store/aio/backends/_graph/monitor.py (223 rows, definitions)
---
```python
class MonitorResult(NamedTuple):
    """Outcome of classifying one monitor poll response.

    ``state`` is the terminal-or-not verdict; ``error`` is the Graph error
    envelope on a ``failed`` poll (``{"code": ..., "message": ...}``), ``None``
    otherwise. ``classified`` is ``False`` when the parser could not find a
    status in the body — a *pending* outcome that the loop tracks separately as
    a ``parse-error`` so the timeout message can distinguish "still running" from
    "kept returning an unreadable body".
    """
    state: MonitorState
    error: dict[str, Any] | None = None
    classified: bool = True
def parse_graph_monitor_response(response: httpx.Response) -> MonitorResult:
    """Classify a Graph monitor poll response into pending / succeeded / failed.

    A completion ``3xx`` redirect is success (the consumer-OneDrive monitor
    answers a finished operation with ``303 See Other`` to the new item, which
    the poll deliberately does not follow). A ``2xx`` body carries a ``status``
    field — ``completed`` is the sole terminal-success value, ``failed`` /
    ``deleteFailed`` are terminal failure (with the ``error`` envelope), and any
    in-flight status (``inProgress`` / ``notStarted`` / ``updating`` / ``waiting``
    / ``deletePending``) is pending. A body with no readable ``status`` is treated
    as pending-but-unclassified so the loop keeps polling rather than failing on a
    single odd response.
    """
async def poll_monitor(
    monitor_url: str,
    client: httpx.AsyncClient,
    *,
    status_parser: Callable[[httpx.Response], MonitorResult] | None = None,
    initial_interval: float | None = None,
    max_interval: float | None = None,
    backoff_factor: float | None = None,
    timeout: float | None = None,
    path: str = "",
    backend: str = BACKEND_NAME,
) -> None:
    """Poll *monitor_url* until the operation finishes; return on success, raise on failure.

    Polls with exponential backoff (``initial_interval`` floor, ``max_interval``
    ceiling, ``backoff_factor`` growth — defaults 1 s / 30 s / 2). A
    ``Retry-After`` header on a poll response raises the wait to at least its
    value. Transient ``5xx`` responses, ``408`` request timeouts, ``429``
    throttles, and transport errors during polling are treated as *pending*, not
    failure. Any other ``4xx`` on the poll request itself (e.g. ``403`` permission
    revoked mid-operation, ``404`` monitor expired/deleted) is **terminal** and
    raises immediately rather than looping until the timeout. ``timeout`` (the backend's
    ``copy_timeout``) bounds the total wall-clock; ``None`` means no ceiling — the
    poll runs until Graph reports a terminal state.

    Raises:
        BackendUnavailable: On ``timeout`` expiry (the message embeds the monitor
            URL, the poll count, and a ``last_status`` token), or mapped from a
            ``failed`` poll whose ``error.code`` is unknown.
        RemoteStoreError: Mapped from a ``failed`` poll body's ``error.code`` via
            the standard table (e.g. ``AlreadyExists`` for ``nameAlreadyExists``),
            or from a terminal ``4xx`` on the poll request via ``status`` + code
            (e.g. ``NotFound`` for ``404``, ``PermissionDenied`` for ``403``).
    """
```

[31/69] src/remote_store/aio/backends/_graph/transfer.py (573 rows, definitions)
---
```python
async def stream_range(
    client: httpx.AsyncClient,
    path: str,
    item: Mapping[str, Any],
    *,
    start: int = 0,
    length: int | None = None,
    refetch: Refetch,
    on_fallback: OnFallback,
    on_range_success: OnRangeSuccess,
    retry: RetryPolicy | None = None,
    backend: str = BACKEND_NAME,
) -> AsyncIterator[bytes]:
    """Stream ``[start, start+length)`` of a file from its download URL.

    ``start=0, length=None`` is a full read and sends no ``Range`` header; any
    other window sends one. Yields the body in chunks, tracking how many it has
    delivered so an interrupted read resumes from the next unread byte:

    * a status-time expiry (``401`` / ``403``) or a connection drop
      (``httpx.TransportError``, possibly mid-body after some chunks have already
      been yielded) re-fetches metadata for a fresh URL, verifies the ``eTag`` is
      unchanged, and re-requests from ``start + delivered`` with a ``Range``
      header — bounded by ``retry.max_attempts``;
    * a drive that ignores ``Range`` (``200`` full entity) or rejects it with a
      non-``416`` ``4xx`` is treated as range-incapable: the full entity is
      re-read into a spool and the requested window served from it
      (*on_fallback* fires once). A ``206`` to a ``Range`` request confirms the
      drive honours ranges and fires *on_range_success* — the backend uses the
      two callbacks to maintain a self-healing, drive-scoped fallback hint.

    Raises:
        BackendUnavailable: Missing download URL, recovery that does not clear
            within the retry budget, an ``eTag`` change mid-read, a non-range
            error status, or a transport failure on the pre-signed host.
        RemoteStoreError: A ``416`` provoked by a malformed (inverted) range.
    """
async def spool_content(content: bytes | AsyncIterator[bytes], *, path: str) -> tuple[BinaryIO, int]:
    """Materialise write *content* into a seekable, replayable reader + its length.

    ``bytes`` is wrapped in an in-memory ``BytesIO`` (no spill). An
    ``AsyncIterator[bytes]`` of unknown length is drained into a
    ``SpooledTemporaryFile`` (system temp, no ``dir=``) so the total can be
    measured before the upload session opens (Graph's ``Content-Range`` needs a
    known total); a spill past the in-memory threshold emits the
    ``graph.upload.spool_spilled`` DEBUG marker. The reader is positioned at
    ``0`` and supports ``seek``/``read`` so chunks replay safely across retries.
    The caller owns closing it.
    """
async def upload_session(
    client: httpx.AsyncClient,
    create_url: str,
    reader: BinaryIO,
    total: int,
    *,
    path: str,
    token_provider: TokenProvider,
    chunk_size: int,
    overwrite: bool,
    retry: RetryPolicy | None,
    on_session_open: Callable[[str], None],
    on_session_close: Callable[[str], None],
    backend: str = BACKEND_NAME,
) -> dict[str, Any]:
    """Upload *reader* (``total`` bytes) via a Graph upload session; return the driveItem.

    Opens a session with ``POST createUploadSession``, then PUTs aligned chunks
    with ``Content-Range: bytes {start}-{end}/{total}``. Each chunk is an
    in-memory byte slice, so the shared retry loop may re-send it on a transient
    ``5xx`` / ``429`` without restarting the session; a ``202`` resumes from the
    server's ``nextExpectedRanges`` rather than the client cursor. Chunk PUTs are
    sent unauthenticated: the session URL is pre-authorised, lives on a different
    host, and carries its own token in the query, so attaching the Graph bearer
    would both leak it cross-host and be rejected.

    On an unrecoverable failure the session is ``DELETE``d best-effort; a ``423``
    is the exception — the session stays valid server-side, so it surfaces as
    ``ResourceLocked`` (carrying the **redacted** session URL — query stripped so
    no credential leaks — and last ``nextExpectedRanges`` for diagnosis) without
    an abort. ``on_session_open`` / ``on_session_close`` register the live session
    URL so the backend's ``close()`` can abort it if a write is mid-flight.
    """
async def abort_upload_session(client: httpx.AsyncClient, session_url: str, *, token_provider: TokenProvider) -> None:
    """Issue a best-effort ``DELETE`` against *session_url*.

    Every failure — a mapped error, a transport drop, a non-2xx status — is
    swallowed to a DEBUG record; cleanup must never propagate.
    """
```

[32/69] src/remote_store/aio/backends/_graph/utils.py (168 rows, definitions)
---
```python
class GraphUtils:
    """Namespace for Graph configuration helpers.

    Carries only ``@staticmethod`` helpers; it is never instantiated. Lives in
    a namespace class rather than on ``GraphBackend`` so it is usable without
    constructing a backend, and so future configuration probes have a home that
    does not crowd the top-level package namespace.
    """
    @staticmethod
    def resolve_drive_id(
        target: str | tuple[str, str] | Mapping[str, str],
        *,
        token_provider: TokenProvider,
        http_client: httpx.AsyncClient | None = None,
        base_url: str = _DEFAULT_BASE_URL,
    ) -> str:
        """Resolve a ``drive_id`` from one of the three target shapes.

        Sync entry point for application wiring; runs ``aresolve_drive_id``
        under a private event loop. See ``aresolve_drive_id`` for the accepted
        shapes and raised errors.
        """
    @staticmethod
    async def aresolve_drive_id(
        target: str | tuple[str, str] | Mapping[str, str],
        *,
        token_provider: TokenProvider,
        http_client: httpx.AsyncClient | None = None,
        base_url: str = _DEFAULT_BASE_URL,
    ) -> str:
        """Resolve a ``drive_id`` from one of three target shapes.

        Accepted shapes:

        - ``"me"`` — the authenticated user's default drive (``GET /me/drive``).
        - a SharePoint site URL ``str`` — the site's default drive; or a
          ``(site_url, library_name)`` tuple — the named document library.
        - a ``{"team_id": ..., "channel_id": ...}`` mapping — a Teams
          channel's backing drive.

        Args:
            target: One of the shapes above.
            token_provider: Bearer-token callable (sync or async).
            http_client: Reuse an existing client; one is created and closed
                per call when omitted.
            base_url: Graph API root.

        Returns:
            The opaque Graph ``drive.id`` string.

        Raises:
            InvalidPath: If ``target`` matches no accepted shape, the SharePoint
                site URL has no host, or the named library does not exist.
            NotFound: If a site/team/channel id resolves but returns ``404``.
            PermissionDenied: If Graph returns ``403`` for the lookup.
            BackendUnavailable: For a transport error, a retryable ``5xx`` /
                ``429`` / ``507``, or a malformed ``@odata.nextLink`` while
                paging a site's document libraries.
        """
```

[33/69] src/remote_store/aio/backends/_memory.py (787 rows, definitions)
---
```python
class AsyncMemoryBackend(AsyncBackend):
    """In-memory async backend using a tree-indexed data structure.

    Zero dependencies, no filesystem access, no network.  Designed as a
    drop-in async backend for unit testing, interactive exploration, and
    documentation examples.

    Supports all capabilities except ``GLOB``.  The full conformance suite
    passes with zero skips.

    Note:
        ``LAZY_READ`` is included here but absent from ``MemoryBackend``
        (the sync mirror).  The sync backend buffers fully in memory; the
        async backend can yield chunks incrementally.  The ``mirrors``
        graph edge between the two does not imply capability parity.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    __mirror__: ClassVar[type[MemoryBackend]] = MemoryBackend
    def __init__(self) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    async def exists(self, path: str) -> bool:
        """Check if a file or folder exists.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if a file or folder exists at *path*.
        """
    async def is_file(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing file.

        Args:
            path: Backend-relative key.

        Returns:
            ``True`` if *path* exists and is a file.
        """
    async def is_folder(self, path: str) -> bool:
        """Return ``True`` if ``path`` is an existing folder.

        Args:
            path: Backend-relative key, or ``""`` for the root.

        Returns:
            ``True`` if *path* exists and is a folder.
        """
    async def read(self, path: str) -> AsyncIterator[bytes]:
        """Open a file for reading and return an async iterator of byte chunks.

        Args:
            path: Backend-relative key.

        Returns:
            An async iterator yielding a single byte chunk (the full content).

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    async def read_bytes(self, path: str) -> bytes:
        """Read the full content of a file as bytes.

        Args:
            path: Backend-relative key.

        Returns:
            The file content.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    async def write(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content to a file.

        Args:
            path: Backend-relative key.
            content: Data to write (bytes or async iterator of bytes).
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-defined string metadata.

        Returns:
            ``WriteResult`` with ``path``, ``size``, ``last_modified``, and
            ``source="native"`` populated.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory.
        """
    async def write_atomic(
        self,
        path: str,
        content: AsyncWritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write content atomically (same as write for in-memory backend).

        Args:
            path: Backend-relative key.
            content: Data to write.
            overwrite: If ``False``, raise if file already exists.
            metadata: Optional user-defined string metadata.

        Returns:
            ``WriteResult`` with ``path``, ``size``, ``last_modified``, and
            ``source="native"`` populated.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If ``path`` names a directory.
        """
    async def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete a file.

        Args:
            path: Backend-relative key.
            missing_ok: If ``True``, do not raise when the file is absent.

        Raises:
            NotFound: If the file is missing and ``missing_ok`` is ``False``.
            InvalidPath: If the path is empty, or if ``path`` names a directory
                (regardless of ``missing_ok``).
        """
    async def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Backend-relative key.
            recursive: If ``True``, delete all contents first.
            missing_ok: If ``True``, do not raise when absent.

        Raises:
            NotFound: If the folder is missing and ``missing_ok`` is ``False``.
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
            InvalidPath: If the path is empty or names an existing file
                (regardless of ``missing_ok``).
        """
    async def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> AsyncIterator[FileInfo]:
        """List files under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.
            recursive: If ``True``, include files in all subdirectories.
            max_depth: Optional maximum folder depth to traverse.

        Returns:
            An async iterator of ``FileInfo`` objects.
        """
    async def list_folders(self, path: str) -> AsyncIterator[FolderEntry]:
        """List immediate subfolders under ``path``.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An async iterator of ``FolderEntry`` objects.
        """
    async def iter_children(self, path: str) -> AsyncIterator[FileInfo | FolderEntry]:
        """Yield both files and folders under ``path`` in a single pass.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            An async iterator of ``FileInfo`` (files) and ``FolderEntry`` (folders).
        """
    async def get_file_info(self, path: str) -> FileInfo:
        """Get metadata for a file.

        Args:
            path: Backend-relative key.

        Returns:
            A ``FileInfo`` with size, modification time, etc.

        Raises:
            InvalidPath: If ``path`` names an existing directory.
            NotFound: If the file does not exist.
        """
    async def get_folder_info(self, path: str) -> FolderInfo:
        """Get metadata for a folder.

        Args:
            path: Backend-relative folder key, or ``""`` for the root.

        Returns:
            A ``FolderInfo`` with file count, total size, etc.

        Raises:
            InvalidPath: If ``path`` names an existing file.
            NotFound: If the folder does not exist.
        """
    async def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename a file.

        ``src == dst`` is a no-op (the file is preserved unchanged).

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            NotFound: If ``src`` does not exist.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and
                ``overwrite`` is ``False``.
            InvalidPath: If ``src`` or ``dst`` is empty, ``src`` names a
                directory, or ``dst`` names an existing directory.
        """
    async def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy a file.

        ``src == dst`` is a no-op (the file is preserved unchanged).

        Args:
            src: Backend-relative source key.
            dst: Backend-relative destination key.
            overwrite: If ``True``, replace any existing file at *dst*.

        Raises:
            NotFound: If ``src`` does not exist.
            AlreadyExists: If ``dst`` exists, ``src != dst``, and
                ``overwrite`` is ``False``.
            InvalidPath: If ``src`` or ``dst`` is empty, ``src`` names a
                directory, or ``dst`` names an existing directory.
        """
    def __repr__(self) -> str:
```

[34/69] src/remote_store/aio/ext/__init__.py (1 rows, definitions)

[35/69] src/remote_store/aio/ext/write.py (65 rows, definitions)
---
```python
async def write_with_hash(
    store: AsyncStore,
    path: str,
    content: bytes | AsyncIterator[bytes],
    *,
    algorithm: str = "sha256",
    overwrite: bool = False,
    metadata: Mapping[str, str] | None = None,
) -> WriteResult:
    """Write *content* and return a ``WriteResult`` with ``digest`` populated.

    Hash is computed client-side as data flows to the backend -- zero extra
    round trips, no buffering for async-iterator input.  ``source`` is
    preserved from the underlying ``store.write()`` result.

    Args:
        store: Target async store.
        path: Destination path.
        content: ``bytes`` or ``AsyncIterator[bytes]``.
        algorithm: ``hashlib`` algorithm name.  Default ``"sha256"``.
        overwrite: Same semantics as ``AsyncStore.write``.
        metadata: Optional user metadata; subject to ``USER_METADATA`` gate.

    Returns:
        ``WriteResult`` with ``digest`` populated from the client-side hash.
    """
```

[36/69] src/remote_store/backends/__init__.py (49 rows, definitions)

[37/69] src/remote_store/backends/_azure.py (1797 rows, definitions)
---
```python
class AzureBackend(Backend):
    """Azure Storage backend.

    Uses the Blob SDK for non-HNS accounts (plain Blob Storage, Azurite) and
    the DataLake SDK for HNS accounts (ADLS Gen2) to get atomic rename and
    real directory support.

    Whether the account is HNS (ADLS Gen2) is declared explicitly via the
    required ``hns`` argument -- the backend does not probe for it.

    ``move()`` on non-HNS accounts is implemented as a server-side copy
    followed by a blob delete.  This is non-atomic: a failure between the
    two steps may leave both source and destination present.  HNS accounts
    use ``rename_file`` which *is* atomic, so an HNS instance could in
    principle advertise ``ATOMIC_MOVE``.  The capability is deliberately
    not declared per-instance: ``CAPABILITIES`` is a single class-level set
    shared by HNS and non-HNS instances alike, so it reports the guarantee
    common to both rather than varying by the declared ``hns`` value.

    Args:
        container: Azure Storage container name (required, non-empty).
        hns: Whether the storage account has Hierarchical Namespace
            enabled (ADLS Gen2). **Required** -- there is no default and
            no runtime auto-detection. Pass ``True`` for ADLS Gen2
            accounts (atomic rename, real directories) or ``False`` for
            flat Blob Storage. Use ``AzureUtils.detect_hns()`` to discover
            the value once if you do not already know it.
        account_name: Storage account name.
        account_url: Full account URL (e.g. ``https://myaccount.dfs.core.windows.net``).
        account_key: Storage account key.
        sas_token: Shared Access Signature token.
        connection_string: Azure Storage connection string.
        credential: Any credential object (e.g. ``DefaultAzureCredential()``).
        client_options: Additional options passed to service clients.
            The library sets ``max_single_put_size``, ``max_block_size``,
            and ``min_large_block_upload_threshold`` defaults for streaming
            memory discipline; user-supplied values take precedence.
        retry: Retry policy for transient failures.
        max_concurrency: Maximum number of parallel connections for
            uploads and downloads (default ``1`` -- sequential).
        reject_write_under_file_ancestor: If ``True``, ``write`` /
            ``write_atomic`` / ``open_atomic`` / ``move`` / ``copy`` HEAD
            each slash-aligned ancestor of the target path on non-HNS
            accounts and raise ``InvalidPath`` on the first regular-file
            hit, matching the cross-backend contract that hierarchical
            filesystems enforce natively. On HNS accounts the kwarg
            short-circuits: ``hdi_isfolder`` rejects the operation
            natively, and the backend detects the file ancestor on that
            rejection and re-raises it as ``InvalidPath``, so HNS delivers
            the cross-backend contract with or without the kwarg set.
            Default ``False``: enabling the check adds one HEAD per
            ancestor per nested-path write; paths without slashes
            short-circuit.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    # M1 (BK-298): close() is terminal — a use-after-close raises
    # BackendUnavailable rather than silently re-initialising. Gates the
    # use-after-close conformance lane.
    close_is_terminal: ClassVar[bool] = True
    def __init__(
        self,
        container: str,
        *,
        hns: bool | None = None,
        account_name: str | None = None,
        account_url: str | None = None,
        account_key: str | Secret | None = None,
        sas_token: str | Secret | None = None,
        connection_string: str | Secret | None = None,
        credential: Any | None = None,
        client_options: dict[str, Any] | None = None,
        retry: RetryPolicy | None = None,
        max_concurrency: int = 1,
        reject_write_under_file_ancestor: bool = False,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def check_health(self) -> None:
        """Verify the backend is reachable and credentials are valid.

        Raises:
            PermissionDenied: If credentials are invalid.
            NotFound: If the container does not exist.
            BackendUnavailable: If the backend cannot be reached.
        """
    def to_key(self, native_path: str) -> str:
    def native_path(self, path: str) -> str:
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with Azure-specific details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="azure"`` and ``details`` containing
            ``container`` and ``account_url``.
        """
    def exists(self, path: str) -> bool:
        """Return ``True`` if a blob or folder exists at *path*; never ``NotFound``.

        Probes the blob first (one HEAD); if absent, probes for a folder (an HNS
        directory, or any blob under the ``path/`` prefix on flat accounts). The
        root always exists.

        Raises:
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing blob (not an HNS directory marker).

        One HEAD round-trip; a missing blob or an ``hdi_isfolder`` directory
        returns ``False``.

        Raises:
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing folder (HNS directory or non-HNS prefix).

        The root is always a folder. Costs one directory HEAD (HNS) or a
        one-item prefix listing (flat).

        Raises:
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a streaming handle.

        Streams the blob body in chunks, so memory stays constant regardless of
        size. On HNS accounts one extra HEAD confirms *path* is a file (not an
        ``hdi_isfolder`` directory) before streaming.

        Raises:
            NotFound: If the blob does not exist.
            InvalidPath: If *path* names a directory.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def read_seekable(self, path: str) -> BinaryIO:
        """Open *path* for random-access reading and return a seekable handle.

        Overrides the base spool: instead of buffering the whole blob, each
        ``read()`` issues one HTTP Range request (``download_blob(offset=,
        length=)``), so only the byte ranges actually read are fetched — ideal
        for Parquet column pruning.

        Raises:
            NotFound: If the blob does not exist.
            InvalidPath: If *path* names a directory.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def read_bytes(self, path: str) -> bytes:
        """Read and return the full blob content as bytes.

        Downloads the whole blob into memory (unlike the lazy ``read`` stream).

        Raises:
            NotFound: If the blob does not exist.
            InvalidPath: If *path* names a directory (HNS).
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* as a blob.

        The content is uploaded with ``upload_blob`` on both flat and HNS
        accounts — a block-blob upload that becomes visible only when its final
        commit succeeds, never as a partially written blob. On flat (non-HNS)
        accounts this is an atomic replace. On hierarchical-namespace (HNS)
        accounts a guaranteed atomic replace is not assured; use ``write_atomic``
        there when readers must never observe an intermediate state.

        Raises:
            AlreadyExists: If the blob exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or (with the
                ``reject_write_under_file_ancestor`` opt-in, or natively on HNS)
                an ancestor exists as a file.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* atomically.

        Readers never observe a partial file. On flat accounts this is the plain
        ``PUT`` (already atomic). On HNS it streams to a hidden temp file and
        promotes it with an atomic ``rename_file`` (the temp file is cleaned up
        on failure).

        Raises:
            AlreadyExists: If the blob exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor exists as a
                file.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    @contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield a writable buffer promoted to *path* atomically on clean exit.

        Writes spool to a temporary file (up to 8 MB in memory, then on disk). On
        flat accounts the buffer is committed with one ``PUT``; on HNS it is
        uploaded to a temp blob and promoted with an atomic ``rename_file``. An
        exception before exit leaves *path* untouched.

        Raises:
            AlreadyExists: If the blob exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor exists as a
                file.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the blob at *path*.

        Raises:
            NotFound: If the blob does not exist (or, on HNS, a path component is
                itself a file) and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a directory (HNS; use ``delete_folder``).
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete a folder.

        Args:
            path: Backend-relative key.
            recursive: If ``True``, delete all contents first.
            missing_ok: If ``True``, do not raise when absent.

        Raises:
            NotFound: If the folder is missing and ``missing_ok`` is ``False``.
            InvalidPath: If ``path`` names a file (use ``delete`` instead).
            DirectoryNotEmpty: If non-empty and ``recursive`` is ``False``.
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield files under *path*, one ``FileInfo`` at a time.

        Lazily pages the service listing (``walk_blobs``/``list_blobs`` on flat
        accounts, ``get_paths`` on HNS); a missing path or a path under a file
        ancestor yields nothing. ``recursive`` lists the whole prefix
        (``max_depth`` prunes client-side).

        Raises:
            PermissionDenied: If credentials are rejected or lack access (401/403),
                surfaced during iteration.
            BackendUnavailable: On throttling (429), 5xx, or transport failure,
                surfaced during iteration.
        """
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """Yield immediate subfolders of *path* as ``FolderEntry`` records.

        One paged prefix listing (``walk_blobs`` common-prefixes on flat
        accounts, non-recursive ``get_paths`` on HNS); a missing path yields
        nothing.

        Raises:
            PermissionDenied: If credentials are rejected or lack access (401/403),
                surfaced during iteration.
            BackendUnavailable: On throttling (429), 5xx, or transport failure,
                surfaced during iteration.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield the immediate files and folders under *path* in one paged listing.

        Overrides the base two-pass default with a single ``walk_blobs`` (flat)
        or ``get_paths`` (HNS) pass, yielding ``FileInfo`` for files and
        ``FolderEntry`` for folders. A missing path yields nothing.

        Raises:
            PermissionDenied: If credentials are rejected or lack access (401/403),
                surfaced during iteration.
            BackendUnavailable: On throttling (429), 5xx, or transport failure,
                surfaced during iteration.
        """
    def glob(self, pattern: str) -> Iterator[FileInfo]:
        """Match files against a glob pattern.

        Args:
            pattern: Glob pattern (e.g., ``"data/*.csv"``, ``"**/*.txt"``).

        Returns:
            An iterator of matching ``FileInfo`` objects.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return file metadata for ``path``.

        Args:
            path: Backend-relative key.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If ``path`` names a directory (HNS: ``hdi_isfolder=true``).
        """
    def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregate metadata for the folder at *path*.

        File count, total size, and latest modification time are gathered by
        paging the whole subtree listing, so cost scales with the number of
        descendants.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If *path* names a file, not a folder.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the file *src* to *dst*.

        On HNS accounts this is a single native ``rename_file`` (atomic). On flat
        accounts it is a server-side copy followed by a delete — not atomic, so a
        failure between the two steps can leave both *src* and *dst* present;
        ``ATOMIC_MOVE`` is therefore not declared. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* names a directory, or an ancestor of
                *dst* exists as a file.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the file *src* to *dst* via a server-side copy.

        Issues ``start_copy_from_url`` so the bytes never pass through the
        client. The copy is not atomic — a failure can leave a partial
        destination. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* names a directory, or an ancestor of
                *dst* exists as a file.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If credentials are rejected or lack access (401/403).
            BackendUnavailable: On throttling (429), 5xx, or transport failure.
        """
    def close(self) -> None:
    def unwrap(self, type_hint: type[T]) -> T:
    def __del__(self) -> None:
    def __repr__(self) -> str:
class AzureUtils:
    """Stateless helpers for Azure Storage accounts.

    A namespace of one-shot utilities that do not require constructing a full
    ``AzureBackend`` -- mirroring ``SFTPUtils`` and ``GraphUtils``.
    """
    @staticmethod
    def detect_hns(
        *,
        account_name: str | None = None,
        account_url: str | None = None,
        account_key: str | Secret | None = None,
        sas_token: str | Secret | None = None,
        connection_string: str | Secret | None = None,
        credential: Any | None = None,
        client_options: dict[str, Any] | None = None,
    ) -> bool:
        """Probe whether a storage account has Hierarchical Namespace enabled.

        Issues a single ``GetAccountInfo`` call against the account, intended
        for discovering the value to pass as ``AzureBackend(hns=...)``. This is
        **fail-loud**: a probe error is mapped to a ``remote_store`` error and
        raised, never swallowed.

        Args:
            account_name: Storage account name.
            account_url: Full account URL (blob endpoint).
            account_key: Storage account key.
            sas_token: Shared Access Signature token.
            connection_string: Azure Storage connection string. When set, takes
                precedence over ``account_url`` / ``account_name``.
            credential: Any credential object (e.g. ``DefaultAzureCredential()``).
            client_options: Additional options passed to the service client.

        Returns:
            ``True`` if Hierarchical Namespace (ADLS Gen2) is enabled, ``False``
            for a flat Blob Storage account.

        Raises:
            RemoteStoreError: If the probe fails (authentication, network, etc.).
        """
    @staticmethod
    async def adetect_hns(
        *,
        account_name: str | None = None,
        account_url: str | None = None,
        account_key: str | Secret | None = None,
        sas_token: str | Secret | None = None,
        connection_string: str | Secret | None = None,
        credential: Any | None = None,
        client_options: dict[str, Any] | None = None,
    ) -> bool:
        """Async sibling of ``detect_hns`` (uses the async Blob SDK).

        Accepts the same parameters as ``detect_hns``.

        Returns:
            ``True`` if Hierarchical Namespace (ADLS Gen2) is enabled,
            ``False`` for a flat Blob Storage account.

        Raises:
            RemoteStoreError: If the probe fails (authentication, network, etc.).
        """
```

[38/69] src/remote_store/backends/_azure_common.py (332 rows, definitions)
---
```python
def validate_azure_params(
    container: str,
    account_name: str | None,
    account_url: str | None,
    connection_string: Any,
    max_concurrency: int,
    hns: bool | None,
) -> None:
    """Validate Azure backend constructor parameters.

    Args:
        container: Azure Storage container name.
        account_name: Storage account name.
        account_url: Full account URL.
        connection_string: Azure Storage connection string (may be ``Secret``).
        max_concurrency: Maximum number of parallel connections.
        hns: Whether Hierarchical Namespace is enabled. Must be declared
            explicitly as a real ``bool`` (``True`` or ``False``); ``None``
            and any non-bool (e.g. the string ``"false"`` from config
            env-var resolution) are rejected.

    Raises:
        ValueError: If any parameter is invalid.
    """
def build_blob_service(
    *,
    connection_string: Any,
    account_url: str | None,
    account_name: str | None,
    credential: Any,
    client_options: Mapping[str, Any] | None = None,
    is_async: bool = False,
) -> Any:
    """Construct a ``BlobServiceClient`` (sync or async) from connection params.

    Shared by the backend's lazy ``_blob_service`` property and
    ``AzureUtils.detect_hns`` so the connection-string / account-URL
    construction lives in one place.

    Args:
        connection_string: Azure Storage connection string (may be ``Secret``).
            When set, takes precedence over ``account_url`` / ``account_name``.
        account_url: Full account URL (blob endpoint).
        account_name: Storage account name; the blob endpoint URL is derived
            from it when ``account_url`` is not given.
        credential: Credential for the URL form (ignored for the
            connection-string form).
        client_options: Extra keyword options for the service client.
        is_async: Build an ``azure.storage.blob.aio`` client instead of the
            synchronous one.

    Returns:
        A ``BlobServiceClient`` instance.

    Raises:
        ValueError: If neither a connection string nor a resolvable URL is given.
    """
def azure_path(path: str) -> str:
    """Normalize path for Azure (strip leading ``/``, collapse double separators).

    Args:
        path: Backend-relative key.

    Returns:
        Normalized Azure blob/path name.
    """
def classify_azure_error(exc: Exception, path: str, backend_name: str) -> RemoteStoreError:
    """Classify an Azure SDK exception into a remote_store error type.

    When called via ``_ErrorMappingStream`` on an ``_AzureRangeReader``,
    the exception may be an ``OSError`` wrapping the original Azure SDK
    exception (via ``__cause__``).  Unwrap before matching.

    Args:
        exc: The Azure SDK exception.
        path: Backend-relative key involved in the operation.
        backend_name: Name of the backend for error reporting.

    Returns:
        An appropriate ``RemoteStoreError`` subclass.
    """
def props_to_fileinfo(props: Any, path: str) -> FileInfo:
    """Convert Azure blob/path properties to ``FileInfo``.

    Args:
        props: Azure SDK blob or path properties object.
        path: Backend-relative key for the file.

    Returns:
        A ``FileInfo`` populated from the properties.
    """
def resolve_credential(
    credential: Any,
    account_key: str | None,
    sas_token: str | None,
    *,
    is_async: bool,
    backend_name: str,
) -> Any:
    """Build credential from constructor parameters.

    Args:
        credential: Explicit credential object, or ``None``.
        account_key: Storage account key.
        sas_token: Shared Access Signature token.
        is_async: If ``True``, import ``DefaultAzureCredential`` from
            ``azure.identity.aio`` instead of ``azure.identity``.
        backend_name: Name of the backend for error reporting.

    Returns:
        A credential suitable for passing to Azure SDK service clients.

    Raises:
        BackendUnavailable: If no credential is provided and ``azure-identity``
            is not installed.
    """
def build_azure_retry(retry: RetryPolicy | None) -> Any | None:
    """Build an Azure ``ExponentialRetry`` from the retry policy, or ``None``.

    Args:
        retry: The retry policy, or ``None`` to skip.

    Returns:
        An ``ExponentialRetry`` instance, or ``None``.
    """
```

[39/69] src/remote_store/backends/_fileinfo.py (24 rows, definitions)
[40/69] src/remote_store/backends/_flat_ns.py (137 rows, definitions)

[41/69] src/remote_store/backends/_http.py (653 rows, definitions)
---
```python
@dataclasses.dataclass
class HttpResponse:
    """Transport-level HTTP response."""
    status: int
    headers: dict[str, str]
    body: BinaryIO
@runtime_checkable
class HttpTransport(Protocol):
    """Internal protocol for pluggable HTTP transports."""
    def get(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
    def head(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
    def close(self) -> None:
class UrllibTransport:
    """HTTP transport using stdlib ``urllib.request``.

    Note:
        Not thread-safe. The shared redirect handler's counter is reset
        per request, so concurrent calls to ``_request()`` would race on
        ``_redirect_count``. urllib openers are not thread-safe in general.
    """
    def __init__(self, *, verify_ssl: bool = True, max_redirects: int = 5) -> None:
    def get(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
        """Send a GET request."""
    def head(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
        """Send a HEAD request."""
    def close(self) -> None:
        """No-op — urllib has no connection pool to close."""
class ReadOnlyHttpBackend(Backend):
    """Read-only backend for HTTP/HTTPS URLs.

    Treats an HTTP endpoint as a file store with
    ``{READ, METADATA, LAZY_READ}`` capabilities (``read()`` streams the
    response body lazily rather than buffering the whole file). Write,
    delete, list, move, and copy operations raise
    ``CapabilityNotSupported``.

    Args:
        base_url: Root URL. A trailing ``/`` is appended if missing.
        headers: Custom headers sent with every request (e.g. API keys).
        timeout: Request timeout in seconds.
        retry: Retry policy for transient errors.
        http_client: Force a specific transport (``"urllib"``, ``"requests"``,
            or ``"httpx"``). Auto-detected if ``None``.
        verify_ssl: Whether to verify TLS certificates.
        max_redirects: Maximum number of redirects to follow.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _CAPABILITIES
    def __init__(
        self,
        base_url: str,
        *,
        headers: dict[str, str] | None = None,
        timeout: float = 30.0,
        retry: RetryPolicy | None = None,
        http_client: str | None = None,
        verify_ssl: bool = True,
        max_redirects: int = 5,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def exists(self, path: str) -> bool:
        """Check existence via HEAD request (falls back to ranged GET)."""
    def is_file(self, path: str) -> bool:
        """HTTP resources are always files."""
    def is_folder(self, path: str) -> bool:
        """HTTP has no folder concept — always returns False."""
    def read(self, path: str) -> BinaryIO:
        """Stream-read a file via GET."""
    def read_bytes(self, path: str) -> bytes:
        """Buffered-read a file via GET."""
    def get_file_info(self, path: str) -> FileInfo:
        """Get file metadata via HEAD request (falls back to ranged GET)."""
    def get_folder_info(self, path: str) -> FolderInfo:
        """HTTP has no folder concept — always raises NotFound."""
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def open_atomic(self, path: str, *, overwrite: bool = False) -> AbstractContextManager[BinaryIO]:
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def check_health(self) -> None:
        """Verify connectivity by sending HEAD to base_url (or GET if HEAD is blocked).

        Note:
            The health check probes ``base_url`` (the root), not a specific
            file.  Many HTTP servers and CDNs return 403 or 404 for directory
            URLs while serving individual files normally.  A failing health
            check therefore does not necessarily mean ``read()`` or
            ``exists()`` will fail on actual file paths.
        """
    def close(self) -> None:
        """Close the underlying transport."""
    def unwrap(self, type_hint: type[T]) -> T:
        """Return the transport if it matches the requested type."""
    def native_path(self, path: str) -> str:
        """Return the full URL for a backend-relative key."""
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with HTTP-specific details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="http"`` and ``details`` containing
            ``url`` and ``method``.
        """
    def to_key(self, native_path: str) -> str:
        """Strip base_url prefix to get a backend-relative key."""
    def __repr__(self) -> str:
```

[42/69] src/remote_store/backends/_http_httpx.py (119 rows, definitions)
---
```python
class HttpxTransport:
    """HTTP transport using ``httpx.Client`` for connection pooling and HTTP/2."""
    def __init__(self, *, verify_ssl: bool = True, max_redirects: int = 5) -> None:
    def get(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
        """Send a streaming GET request."""
    def head(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
        """Send a HEAD request."""
    def close(self) -> None:
        """Close the httpx client."""
```

[43/69] src/remote_store/backends/_http_requests.py (87 rows, definitions)
---
```python
class RequestsTransport:
    """HTTP transport using ``requests.Session`` for connection pooling."""
    def __init__(self, *, verify_ssl: bool = True, max_redirects: int = 5) -> None:
    def get(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
        """Send a GET request."""
    def head(self, url: str, headers: dict[str, str], timeout: float) -> HttpResponse:
        """Send a HEAD request."""
    def close(self) -> None:
        """Close the requests session."""
```

[44/69] src/remote_store/backends/_local.py (795 rows, definitions)
---
```python
class LocalBackend(Backend):
    """Local filesystem backend using only the Python standard library.

    ``move()`` uses ``shutil.move``, which calls ``os.rename`` for
    same-filesystem moves (atomic) but falls back to copy-then-delete
    for cross-filesystem moves (not atomic).  ``ATOMIC_MOVE`` is
    declared because within-root moves are always same-filesystem.

    Args:
        root: Absolute path to the root directory on the local filesystem.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    def __init__(self, root: str) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def check_health(self) -> None:
        """Confirm the root directory exists and is readable.

        A lightweight two-syscall probe (``exists`` then ``os.access``) that
        touches no file content and is safe to call repeatedly.

        Raises:
            NotFound: If the root directory does not exist.
            PermissionDenied: If the root directory exists but is not readable.
        """
    def to_key(self, native_path: str) -> str:
    def native_path(self, path: str) -> str:
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with local filesystem details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="local"`` and ``details`` containing
            ``root`` and ``absolute_path``.
        """
    def exists(self, path: str) -> bool:
        """Return ``True`` if a file or folder exists at *path*; never ``NotFound``.

        Returns ``False`` when a path component is itself a file, since the
        directory descent cannot proceed.

        Raises:
            InvalidPath: If *path* escapes the backend root.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing regular file.

        ``False`` if *path* is absent, names a directory, or has a file as an
        ancestor component.

        Raises:
            InvalidPath: If *path* escapes the backend root.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing directory.

        ``False`` if *path* is absent, names a file, or has a file as an ancestor
        component.

        Raises:
            InvalidPath: If *path* escapes the backend root.
        """
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a lazy binary stream.

        The stream reads on demand from the open file handle, so memory stays
        constant regardless of file size.

        Raises:
            NotFound: If the file does not exist, or a path component is itself a
                file (the directory descent fails with ``ENOTDIR``).
            InvalidPath: If *path* names a directory.
            PermissionDenied: If the OS denies read access to the file.
        """
    def read_bytes(self, path: str) -> bytes:
        """Read and return the full file content as bytes.

        Materialises the whole file in memory (unlike the lazy ``read`` stream),
        so size is bounded by available memory.

        Raises:
            NotFound: If the file does not exist, or a path component is itself a
                file (``ENOTDIR`` on descent).
            InvalidPath: If *path* names a directory.
            PermissionDenied: If the OS denies read access to the file.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path*, creating parent directories as needed.

        The bytes land directly at the final path (no temp-and-rename), so a
        crash or error mid-write can leave a partial or truncated file there —
        use ``write_atomic`` when readers must never observe a half-written file.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor of *path*
                exists as a regular file.
            PermissionDenied: If the OS denies write access.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* atomically via a temp file plus ``os.replace``.

        Readers never observe a partial file: the body is written to a temporary
        file in the destination directory and atomically renamed over *path* on
        success (the temp file is removed on error). The rename is atomic because
        the temp file shares *path*'s directory, hence its filesystem.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor of *path*
                exists as a regular file.
            PermissionDenied: If the OS denies write access.
        """
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield a writable stream promoted to *path* atomically on clean exit.

        Writes go to a temp file in *path*'s directory; on normal exit it is
        atomically renamed over *path* (``os.replace``), and on any exception the
        temp file is removed and *path* is left untouched. Readers therefore
        never see a partially written file.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor of *path*
                exists as a regular file.
            PermissionDenied: If the OS denies write access.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the file at *path*.

        Raises:
            NotFound: If the file does not exist (or a path component is itself a
                file) and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a directory — a type mismatch that
                ``missing_ok`` does not silence (use ``delete_folder``).
            PermissionDenied: If the OS denies the unlink.
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete the folder at *path*.

        ``recursive=True`` removes the folder and its whole subtree
        (``shutil.rmtree``); this is not atomic — an error partway through can
        leave the tree partially deleted. ``recursive=False`` removes only an
        empty folder (``rmdir``).

        Raises:
            NotFound: If the folder does not exist and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a file, not a folder.
            DirectoryNotEmpty: If the folder is non-empty and ``recursive`` is
                ``False``.
            PermissionDenied: If the OS denies removal.
        """
    def glob(self, pattern: str) -> Iterator[FileInfo]:
        """Yield files matching *pattern* under the root, one at a time.

        Lazily walks ``Path.glob``; entries that escape the root through a
        symlink are skipped rather than returned, and directories are omitted —
        only files are yielded.
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield files under *path*, one ``FileInfo`` at a time.

        Lazy: entries are produced as the directory is scanned, so listing a
        large tree holds only the current entry in memory. A missing or
        non-folder *path* yields nothing (no error). With ``recursive`` and
        ``max_depth`` set, traversal is pruned at the depth bound during the
        ``os.walk`` rather than filtered afterwards.
        """
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """Yield immediate subfolders of *path* as ``FolderEntry`` records.

        Lazy single-level scan; a missing or non-folder *path* yields nothing.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield the immediate files and folders under *path* in one scan.

        Overrides the base (which chains ``list_files`` and ``list_folders``,
        two passes) to walk the directory once, yielding ``FileInfo`` for files
        and ``FolderEntry`` for folders. A missing or non-folder *path* yields
        nothing.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return metadata for the file at *path* from a single ``stat``.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* names a directory, not a file.
        """
    def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregate metadata for the folder at *path*.

        File count, total size, and latest modification time are computed by
        walking the entire subtree and ``stat``-ing every file, so cost is
        O(files-in-subtree) — not a constant-time lookup.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If *path* names a file, not a folder.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the file *src* to *dst*.

        Atomic when *src* and *dst* are on the same filesystem (``os.rename`` via
        ``shutil.move``) — which within a single root they always are, so
        ``ATOMIC_MOVE`` is declared. A cross-filesystem move (e.g. a bind-mounted
        subtree) falls back to copy-then-delete, which is not atomic. ``src ==
        dst`` is a no-op; parent directories of *dst* are created as needed.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* names a directory, or an ancestor of
                *dst* exists as a regular file.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If the OS denies the operation.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the file *src* to *dst*, preserving mode and mtime (``copy2``).

        Not atomic: content is streamed to *dst*, so a crash mid-copy can leave a
        partial file at the destination. ``src == dst`` is a no-op; parent
        directories of *dst* are created as needed.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* names a directory, or an ancestor of
                *dst* exists as a regular file.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If the OS denies the operation.
        """
    def __repr__(self) -> str:
```

[45/69] src/remote_store/backends/_memory.py (752 rows, definitions)
---
```python
class MemoryBackend(Backend):
    """In-memory backend using a tree-indexed data structure.

    Zero dependencies, no filesystem access, no network.  Designed as a
    drop-in backend for unit testing, interactive exploration, and
    documentation examples.

    All capabilities except ``GLOB`` are supported.  The full conformance
    suite passes with zero skips.

    Every mutating operation runs under a single process-wide lock, so
    ``write``, ``write_atomic``, ``move``, and ``copy`` are atomic with respect
    to concurrent callers in the same process (``ATOMIC_MOVE`` is advertised).
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    def __init__(self) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def exists(self, path: str) -> bool:
        """Return ``True`` if a file or folder exists at *path*; never ``NotFound``.

        The root (``""``) always exists.

        Raises:
            InvalidPath: If *path* is absolute or contains a ``..`` segment.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing file (``False`` for the root or a folder).

        Raises:
            InvalidPath: If *path* is absolute or contains a ``..`` segment.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing folder; the root is always a folder.

        Raises:
            InvalidPath: If *path* is absolute or contains a ``..`` segment.
        """
    def read(self, path: str) -> BinaryIO:
        """Return a binary stream over the stored bytes for *path*.

        The value materialises in memory rather than streaming (``LAZY_READ`` is
        not advertised): the returned stream wraps a copy of the resident bytes,
        so peak memory scales with the object size. Access is guarded by a single
        process-wide lock.

        Raises:
            NotFound: If no file exists at *path*.
            InvalidPath: If *path* names a folder.
        """
    def read_bytes(self, path: str) -> bytes:
        """Return the full stored content of *path* as bytes.

        Copies the resident value out under the lock; like ``read`` it holds the
        whole object in memory.

        Raises:
            NotFound: If no file exists at *path*.
            InvalidPath: If *path* names a folder.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Store *content* at *path*, creating parent folders implicitly.

        The whole body is buffered into memory before the store (streams are
        drained fully first — no lazy write), and the swap-in happens under a
        single process-wide lock, so the write is atomic with respect to
        concurrent callers in the same process.

        Raises:
            AlreadyExists: If a file exists at *path* and ``overwrite`` is ``False``.
            InvalidPath: If *path* is empty or names a folder, or an ancestor of
                *path* exists as a file.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Store *content* at *path* atomically (delegates to ``write``).

        Memory writes are already atomic under the process-wide lock, so this is
        exactly ``write``; the whole body is buffered first.

        Raises:
            AlreadyExists: If a file exists at *path* and ``overwrite`` is ``False``.
            InvalidPath: If *path* is empty or names a folder, or an ancestor of
                *path* exists as a file.
        """
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield an in-memory buffer committed to *path* atomically on clean exit.

        Writes accumulate in a ``BytesIO``; on exit the buffer is stored via
        ``write`` under the process-wide lock, so *path* updates in one atomic
        step. An exception before exit leaves *path* untouched.

        Raises:
            AlreadyExists: If a file exists at *path* and ``overwrite`` is ``False``.
            InvalidPath: If *path* is empty, or (on commit) names a folder or has
                a file ancestor.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the file at *path*.

        Raises:
            NotFound: If no file exists at *path* and ``missing_ok`` is ``False``.
            InvalidPath: If *path* is empty or names a folder (a type mismatch
                ``missing_ok`` does not silence).
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete the folder at *path*.

        ``recursive=True`` detaches the whole subtree in one locked step;
        ``recursive=False`` removes only an empty folder.

        Raises:
            NotFound: If no folder exists at *path* and ``missing_ok`` is ``False``.
            InvalidPath: If *path* is empty or names a file.
            DirectoryNotEmpty: If the folder is non-empty and ``recursive`` is
                ``False``.
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield files under *path*.

        A missing or non-folder *path* yields nothing. The tree is snapshotted
        under the lock and iterated outside it, so a long listing does not hold
        the lock; ``recursive`` walks the whole subtree (``max_depth`` prunes it).
        """
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """Yield immediate subfolders of *path* as ``FolderEntry`` records.

        A missing or non-folder *path* yields nothing; children are snapshotted
        under the lock and yielded outside it.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield the immediate files and folders under *path* in one snapshot.

        Overrides the base two-pass default: takes one locked snapshot of the
        directory and yields ``FileInfo`` for files and ``FolderEntry`` for
        folders. A missing or non-folder *path* yields nothing.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return metadata for the file at *path* from the in-memory node.

        Raises:
            NotFound: If no file exists at *path* (including the empty path).
            InvalidPath: If *path* names a folder.
        """
    def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregate metadata for the folder at *path*.

        File count, total size, and latest modification time are computed by
        walking the whole subtree, so cost scales with the number of descendants.

        Raises:
            NotFound: If no folder exists at *path*.
            InvalidPath: If *path* names a file.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the file *src* to *dst* by re-linking the tree node.

        Detach-from-source and attach-to-destination happen in one locked step,
        so the move is atomic with respect to concurrent callers (``ATOMIC_MOVE``
        is advertised) and the payload is never copied. ``src == dst`` verifies
        the source is a file and is otherwise a no-op.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* is empty, *src* names a folder, or
                *dst* names an existing folder.
            AlreadyExists: If *dst* is an existing file and ``overwrite`` is
                ``False``.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the file *src* to *dst* with an independent byte buffer.

        The destination receives a fresh ``bytearray`` copy of the source bytes,
        all in one locked step (atomic). ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* is empty, *src* names a folder, or
                *dst* names an existing folder.
            AlreadyExists: If *dst* is an existing file and ``overwrite`` is
                ``False``.
        """
    def __repr__(self) -> str:
```

[46/69] src/remote_store/backends/_memory_tree.py (48 rows, definitions)
---
```python
@dataclass(slots=True)
class FileEntry:
    """Mutable file node in the in-memory tree."""
    data: bytearray
    modified_at: datetime
    content_type: str | None = None
    metadata: dict[str, str] | None = None
@dataclass(slots=True)
class DirNode:
    """Directory node — children map names to files or subdirectories."""
    children: dict[str, DirNode | FileEntry] = field(default_factory=dict)
class FileSnapshot:
    """Frozen copy of ``FileEntry`` scalars for lock-free iteration.

    Copies size, modified_at, and content_type at snapshot time so that
    concurrent writes to the live ``FileEntry`` cannot produce inconsistent
    ``FileInfo`` objects (e.g. new size with old timestamp).
    """
    __slots__ = ("size", "modified_at", "content_type", "metadata")
    def __init__(self, entry: FileEntry) -> None:
```

[47/69] src/remote_store/backends/_s3.py (507 rows, definitions)
---
```python
class S3Backend(_S3Base):
    """S3-compatible object storage backend using s3fs.

    ``move()`` is implemented as a server-side copy followed by a delete.
    This is non-atomic: a crash or network error between the two steps may
    leave both source and destination present.  ``ATOMIC_MOVE`` is not
    declared.

    Args:
        bucket: S3 bucket name (required, non-empty).
        endpoint_url: Custom endpoint URL (e.g. for MinIO).
        key: AWS access key ID.
        secret: AWS secret access key.
        region_name: AWS region name.
        tls_ca_bundle: Path to a PEM CA bundle file.  Falls back to
            ``AWS_CA_BUNDLE`` / ``REQUESTS_CA_BUNDLE`` / ``SSL_CERT_FILE``.
        client_options: Additional options passed to s3fs.
        reject_write_under_file_ancestor: If ``True``, ``write`` /
            ``write_atomic`` / ``open_atomic`` / ``move`` / ``copy`` HEAD
            each slash-aligned ancestor of the target path and raise
            ``InvalidPath`` on the first regular-file hit, matching the
            cross-backend contract that hierarchical filesystems enforce
            natively. Default ``False``: each nested-path write otherwise
            pays one HEAD per ancestor; paths without slashes
            short-circuit.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    def __init__(
        self,
        bucket: str,
        *,
        endpoint_url: str | None = None,
        key: str | Secret | None = None,
        secret: str | Secret | None = None,
        region_name: str | None = None,
        tls_ca_bundle: str | None = None,
        client_options: dict[str, Any] | None = None,
        retry: RetryPolicy | None = None,
        reject_write_under_file_ancestor: bool = False,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def check_health(self) -> None:
        """Confirm the bucket is reachable and credentials valid via one ``HeadBucket``.

        Raises:
            NotFound: If the bucket does not exist.
            PermissionDenied: If the credentials are rejected or lack access.
            BackendUnavailable: On a transport or service failure, or after
                ``close()``.
        """
    def native_path(self, path: str) -> str:
    def exists(self, path: str) -> bool:
        """Return ``True`` if an object or prefix exists at *path*; never ``NotFound``.

        Raises:
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing object (``False`` if absent or a prefix).

        Raises:
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing virtual folder (a common prefix).

        Raises:
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a streaming handle.

        s3fs reads the object lazily in range-backed chunks, so memory stays
        constant regardless of size.

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def read_bytes(self, path: str) -> bytes:
        """Read and return the full object content as bytes.

        Downloads the whole object into memory (unlike the lazy ``read`` stream).

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* as an S3 object.

        The upload commits atomically — a reader sees either the old object or the
        new one, never a partial. A streamed write that fails mid-body aborts the
        multipart upload rather than finalising a truncated object, so no partial
        object is ever left at *path*.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the ``reject_write_under_file_ancestor`` opt-in, if
                an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* atomically (delegates to ``write``).

        An S3 ``PUT`` is already atomic, so this is exactly ``write``.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    @contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield a writable buffer committed to *path* atomically on clean exit.

        Writes spool to a temporary file (up to 8 MB in memory, then on disk); on
        clean exit the buffer is uploaded in a single atomic ``PUT``. An exception
        before exit leaves *path* untouched.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the object at *path*.

        Raises:
            NotFound: If the object does not exist and ``missing_ok`` is ``False``.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete the virtual folder at *path*.

        ``recursive=True`` removes every object under the prefix; this is a
        best-effort multi-object delete, not atomic, so an interruption can leave
        the prefix partially deleted. ``recursive=False`` removes the prefix only
        when it has no contents.

        Raises:
            NotFound: If no object exists under *path* and ``missing_ok`` is ``False``.
            DirectoryNotEmpty: If the prefix is non-empty and ``recursive`` is ``False``.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return metadata for the object at *path* from one ``HeadObject``.

        The HEAD is issued with ``ChecksumMode=ENABLED`` so a stored checksum
        surfaces as the ``FileInfo`` digest.

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the object *src* to *dst*.

        Implemented as a server-side copy followed by a delete of *src*. This is
        not atomic — a crash or network error between the two steps can leave both
        *src* and *dst* present — so ``ATOMIC_MOVE`` is not declared. ``src == dst``
        is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *dst* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the object *src* to *dst* via a server-side ``CopyObject``.

        The bytes are copied entirely server-side (never through the client).
        Like ``move``, the operation carries no cross-operation atomicity
        guarantee. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *dst* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def close(self) -> None:
    def unwrap(self, type_hint: type[T]) -> T:
    def __repr__(self) -> str:
```

[48/69] src/remote_store/backends/_s3_base.py (619 rows, definitions)

[49/69] src/remote_store/backends/_s3_boto3.py (1016 rows, definitions)
---
```python
class S3Boto3Backend(Backend):
    """S3-compatible backend using boto3 directly (no s3fs / pyarrow).

    Drop-in alternative to ``S3Backend`` with the same constructor signature.
    ``move()`` is a server-side ``copy_object`` followed by ``delete_object``
    and is non-atomic, so ``ATOMIC_MOVE`` is not declared.

    Args:
        bucket: S3 bucket name (required, non-empty).
        endpoint_url: Custom endpoint URL (e.g. for MinIO). When set, the
            client uses path-style addressing for emulator compatibility.
        key: AWS access key ID.
        secret: AWS secret access key.
        region_name: AWS region name.
        tls_ca_bundle: Path to a PEM CA bundle file. Falls back to
            ``AWS_CA_BUNDLE`` / ``REQUESTS_CA_BUNDLE`` / ``SSL_CERT_FILE``.
        client_options: Optional dict; a ``config_kwargs`` entry is merged
            into the ``botocore.config.Config`` (PoC-minimal passthrough).
        retry: Retry policy; only ``max_attempts`` maps to botocore.
        reject_write_under_file_ancestor: If ``True``, ``write`` /
            ``write_atomic`` / ``open_atomic`` / ``move`` / ``copy`` HEAD each
            slash-aligned ancestor of the target and raise ``InvalidPath`` on
            the first regular-file hit (flat-namespace opt-in, default off).
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    # M1 (BK-298): close() is terminal — a use-after-close raises
    # BackendUnavailable rather than silently re-creating the boto3 client.
    close_is_terminal: ClassVar[bool] = True
    def __init__(
        self,
        bucket: str,
        *,
        endpoint_url: str | None = None,
        key: str | Secret | None = None,
        secret: str | Secret | None = None,
        region_name: str | None = None,
        tls_ca_bundle: str | None = None,
        client_options: dict[str, Any] | None = None,
        retry: RetryPolicy | None = None,
        reject_write_under_file_ancestor: bool = False,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def native_path(self, path: str) -> str:
    def to_key(self, native_path: str) -> str:
    def resolve(self, path: str) -> ResolutionPlan:
    def exists(self, path: str) -> bool:
        """Return ``True`` if an object or prefix exists at *path*; never ``NotFound``.

        One HEAD, falling back to a one-key prefix listing. The root always exists.

        Raises:
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after
                ``close()``.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing object (one HEAD).

        Raises:
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing virtual folder (a common prefix).

        A same-named object shadows the prefix (flat namespace), so a file returns
        ``False``. The root is always a folder.

        Raises:
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a streaming handle.

        Reads lazily via ranged ``GetObject`` requests wrapped in a
        ``BufferedReader``, so memory stays constant regardless of size.

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def read_seekable(self, path: str) -> BinaryIO:
    def read_bytes(self, path: str) -> bytes:
        """Read and return the full object content as bytes (one ``GetObject``).

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* as an S3 object.

        The upload commits atomically: a ``bytes`` payload is a single ``PUT``,
        and a streamed payload uploads multipart and only becomes visible on
        completion — a mid-stream failure aborts the upload, so no truncated
        object is ever left at *path*.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the ``reject_write_under_file_ancestor`` opt-in, if
                an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* atomically (delegates to ``write``).

        A ``PUT`` and a completed multipart upload are both atomic and
        ``upload_fileobj`` aborts on failure, so ``write`` is already atomic-safe
        and no pre-buffering is needed.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    @contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield a writable buffer committed to *path* atomically on clean exit.

        Writes spool to a temporary file (up to 8 MB in memory, then on disk) and
        upload on clean exit, so *path* never holds a partial object. An exception
        before exit leaves *path* untouched.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the object at *path*.

        Raises:
            NotFound: If the object does not exist and ``missing_ok`` is ``False``.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete the virtual folder at *path*.

        ``recursive=True`` deletes every object under the prefix; this is a
        best-effort multi-object delete, not atomic, so an interruption can leave
        the prefix partially deleted. ``recursive=False`` refuses a non-empty
        prefix.

        Raises:
            NotFound: If no object exists under *path* and ``missing_ok`` is ``False``.
            DirectoryNotEmpty: If the prefix is non-empty and ``recursive`` is ``False``.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return metadata for the object at *path* from one ``HeadObject``.

        Issued with ``ChecksumMode=ENABLED`` so a stored checksum surfaces as the
        ``FileInfo`` digest.

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregate metadata for the virtual folder *path*.

        File count, total size, and latest modification time come from paging the
        whole prefix listing, so cost scales with the number of descendants.

        Raises:
            NotFound: If no object exists under *path*.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield files under *path*, one ``FileInfo`` at a time.

        Lazily pages a delimiter-scoped ``ListObjectsV2`` breadth-first
        (non-recursive is depth 0; ``recursive`` with ``max_depth`` bounds the
        descent). A missing prefix yields nothing. Listings are strongly
        consistent — a just-written object appears in a later listing.
        """
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """Yield immediate subfolders of *path* as ``FolderEntry`` records.

        One delimiter-scoped ``ListObjectsV2`` returning common prefixes; a
        missing prefix yields nothing.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield the immediate files and folders under *path* in one paged listing.

        Overrides the base two-pass default with a single delimiter-scoped
        ``ListObjectsV2``, yielding ``FileInfo`` for objects and ``FolderEntry``
        for common prefixes. A missing prefix yields nothing.
        """
    def glob(self, pattern: str) -> Iterator[FileInfo]:
        """Yield files whose key matches the glob *pattern*.

        Narrows to the pattern's literal prefix, lists that subtree, and applies
        the full glob regex to each key.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the object *src* to *dst*.

        A server-side ``CopyObject`` followed by a ``DeleteObject`` on *src*. This
        is not atomic — a failure between the two steps can leave both present —
        so ``ATOMIC_MOVE`` is not declared. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *dst* exists as an object.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the object *src* to *dst* via a server-side ``CopyObject``.

        The bytes are copied entirely server-side. Like ``move``, the operation
        carries no cross-operation atomicity guarantee. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *dst* exists as an object.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def check_health(self) -> None:
        """Confirm the bucket is reachable and credentials valid via one ``HeadBucket``.

        Raises:
            NotFound: If the bucket does not exist.
            PermissionDenied: If the credentials are rejected or lack access (403).
            BackendUnavailable: On throttling, 5xx, or transport failure, or after ``close()``.
        """
    def close(self) -> None:
    def unwrap(self, type_hint: type[T]) -> T:
    def __repr__(self) -> str:
```

[50/69] src/remote_store/backends/_s3_pyarrow.py (610 rows, definitions)
---
```python
class S3PyArrowBackend(_S3Base):
    """Hybrid S3 backend: PyArrow for reads/writes/copies, s3fs for listing/metadata.

    Drop-in alternative to ``S3Backend`` with the same constructor signature.
    Uses PyArrow's C++ S3 filesystem for data-path operations (higher throughput
    for large files) and s3fs for control-path operations (listing, metadata,
    deletion).

    ``move()`` is implemented as a PyArrow copy followed by an s3fs delete.
    This is non-atomic: a crash or network error between the two steps may
    leave both source and destination present.  ``ATOMIC_MOVE`` is not
    declared.

    Args:
        bucket: S3 bucket name (required, non-empty).
        endpoint_url: Custom endpoint URL (e.g. for MinIO).
        key: AWS access key ID.
        secret: AWS secret access key.
        region_name: AWS region name.
        tls_ca_bundle: Path to a PEM CA bundle file.  Falls back to
            ``AWS_CA_BUNDLE`` / ``REQUESTS_CA_BUNDLE`` / ``SSL_CERT_FILE``.
        client_options: Additional options passed to s3fs.
        reject_write_under_file_ancestor: If ``True``, ``write`` /
            ``write_atomic`` / ``open_atomic`` / ``move`` / ``copy`` HEAD
            each slash-aligned ancestor of the target path and raise
            ``InvalidPath`` on the first regular-file hit, matching the
            cross-backend contract that hierarchical filesystems enforce
            natively. Default ``False``: each nested-path write otherwise
            pays one HEAD per ancestor; paths without slashes
            short-circuit.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    def __init__(
        self,
        bucket: str,
        *,
        endpoint_url: str | None = None,
        key: str | Secret | None = None,
        secret: str | Secret | None = None,
        region_name: str | None = None,
        tls_ca_bundle: str | None = None,
        client_options: dict[str, Any] | None = None,
        retry: RetryPolicy | None = None,
        reject_write_under_file_ancestor: bool = False,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def check_health(self) -> None:
    def native_path(self, path: str) -> str:
    def exists(self, path: str) -> bool:
    def is_file(self, path: str) -> bool:
    def is_folder(self, path: str) -> bool:
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a streaming handle.

        Uses PyArrow's ``open_input_file`` (higher throughput for large objects)
        rather than the s3fs reader; still lazy, so memory stays constant.

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def read_bytes(self, path: str) -> bytes:
        """Read and return the full object content as bytes (via PyArrow).

        Raises:
            NotFound: If the object does not exist.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path*, streaming straight to a multipart upload.

        Unlike ``S3Backend.write``, a plain streamed write here is **not** atomic:
        PyArrow's output stream exposes no abort, so a failure mid-body finalises
        a *truncated* object at *path*. Use ``write_atomic`` when readers must
        never observe a partial object. A ``bytes`` payload (no streaming) commits
        in one shot.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the ``reject_write_under_file_ancestor`` opt-in, if
                an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* atomically by buffering before upload.

        The whole body is buffered first (a ``bytes`` payload is already
        materialised and delegates straight through), so a source failure happens
        off the wire and leaves no object at *path* — closing the atomicity gap in
        the plain streaming ``write``.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    @contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield a writable buffer committed to *path* atomically on clean exit.

        Writes spool to a temporary file (up to 8 MB in memory, then on disk) and
        upload only on clean exit, so *path* never holds a partial object. An
        exception before exit leaves *path* untouched.

        Raises:
            AlreadyExists: If the object exists and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *path* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def get_file_info(self, path: str) -> FileInfo:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the object *src* to *dst*.

        Existence checks and the delete go through s3fs; the copy is a PyArrow
        ``copy_file``. Copy-then-delete is not atomic — a failure between the two
        steps can leave both *src* and *dst* present — so ``ATOMIC_MOVE`` is not
        declared. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *dst* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the object *src* to *dst* via a PyArrow ``copy_file``.

        Like ``move``, the operation carries no cross-operation atomicity
        guarantee. ``src == dst`` is a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is ``False``.
            InvalidPath: With the opt-in, if an ancestor of *dst* exists as an object.
            PermissionDenied: If the credentials lack access.
            BackendUnavailable: On a transport or service failure, or after ``close()``.
        """
    def close(self) -> None:
    def unwrap(self, type_hint: type[T]) -> T:
    def __repr__(self) -> str:
```

[51/69] src/remote_store/backends/_sftp.py (1856 rows, definitions)
---
```python
class HostKeyPolicy(Enum):
    """Controls how unknown remote host keys are handled.

    Attributes:
        STRICT: Reject unknown hosts (production default).
        TRUST_ON_FIRST_USE: Save on first connect, verify after.
        AUTO_ADD: Accept any key (dev/testing ONLY).

    Accepts the enum-name forms (``"auto_add"``, ``"trust_on_first_use"``,
    ``"STRICT"``) in addition to the canonical value strings (``"auto"``,
    ``"tofu"``, ``"strict"``).
    """
    STRICT = "strict"
    TRUST_ON_FIRST_USE = "tofu"
    AUTO_ADD = "auto"
class SFTPUtils:
    """SFTP setup utilities for key loading and host verification.

    Groups helpers that assist with SFTP backend configuration:

    - ``SFTPUtils.load_private_key(...)`` -- load RSA keys from file or PEM string
    - ``SFTPUtils.HostKeyPolicy`` -- enum controlling unknown host key behavior
    - ``SFTPUtils.scan_host_keys(host, port=22)`` -- preflight host-key
      discovery; returns a ``known_hosts``-formatted line for committing
      into a ``host.keys`` file
    - ``SFTPUtils.scan_host_algorithms(host, port=22)`` -- raw-socket SSH
      KEXINIT probe; returns the server's algorithm advertisement
      (kex / host-key / cipher / MAC / compression name-lists) for
      diagnosing ``IncompatiblePeer`` failures
    - ``SFTPUtils.enable_ssh_rsa_compat()`` -- restore ``ssh-rsa`` (SHA-1)
      acceptance for legacy SFTP servers (see method docstring for the
      security tradeoff)

    !!! example

        ```python
        from remote_store.backends import SFTPUtils, SFTPBackend

        key = SFTPUtils.load_private_key("~/.ssh/id_rsa", from_file=True)
        backend = SFTPBackend(
            host="sftp.example.com",
            pkey=key,
            host_key_policy=SFTPUtils.HostKeyPolicy.AUTO_ADD,
        )
        ```
    """
    HostKeyPolicy = HostKeyPolicy
    @staticmethod
    def load_private_key(source: str, *, from_file: bool = False) -> Any:  # pragma: no cover
        """Load an RSA private key from a file path or a PEM string.

        Args:
            source: File path (if ``from_file=True``) or PEM-encoded string.
            from_file: If ``True``, treat *source* as a file path.

        Returns:
            ``paramiko.RSAKey``
        """
    @staticmethod
    def scan_host_keys(host: str, port: int = 22, *, timeout: float = 10.0) -> str:
        """Discover an SFTP server's host key without authenticating.

        Opens a ``paramiko.Transport`` to *host*:*port*, performs key exchange,
        captures the server's offered host key, closes the connection, and
        returns the key as a single ``known_hosts``-formatted line. No
        authentication is attempted; only the SSH key-exchange handshake runs.

        Use this to populate a committed ``host.keys`` file for production
        `STRICT` policy use, without going through a TOFU connect first.

        Args:
            host: Hostname or IP address of the SFTP server.
            port: SSH port (default: 22).
            timeout: Socket and KEX timeout in seconds (default: 10).

        Returns:
            A single ``known_hosts``-format line:
            ``"<host_label> <key_type> <base64_key>"``. Per OpenSSH convention,
            *host_label* is ``"[host]:port"`` when *port* is not 22, and the
            bare hostname otherwise. The trailing newline is not included.

            Returns only the **negotiated** key for one handshake (whichever
            key type paramiko picked: usually one of ed25519, ecdsa, rsa),
            not every key the server offers. ``ssh-keyscan`` returns one line
            per offered type by default; this helper does not. If the server
            offers multiple key types and paramiko later negotiates a
            different one than the pinned line, the connection fails with
            ``BadHostKeyException``. Callers that need full-type coverage
            must call this helper multiple times under different
            ``disabled_algorithms`` settings to force each type in turn.

        Raises:
            paramiko.SSHException: Negotiation failed (e.g. legacy server
                offering only ``ssh-rsa``; call ``enable_ssh_rsa_compat()``
                first if so).
            OSError: Socket-level failure (host unreachable, port refused,
                DNS error, timeout).

        !!! example

            ```python
            from pathlib import Path

            from remote_store.backends import SFTPUtils

            entry = SFTPUtils.scan_host_keys("sftp.example.com")
            Path("host.keys").write_text(entry + "\\n")
            ```
        """
    @staticmethod
    def scan_host_algorithms(
        host: str,
        port: int = 22,
        *,
        timeout: float = 10.0,
    ) -> dict[str, list[str] | str]:
        """Discover an SFTP server's algorithm advertisement without authenticating.

        Opens a raw TCP socket, exchanges SSH banners, reads the server's
        first ``SSH_MSG_KEXINIT`` packet (RFC 4253 § 7.1), parses the ten
        name-lists it carries, and returns them as a dictionary. No
        paramiko, no key exchange completes, no authentication is attempted.

        Use this to identify the failure shape behind an
        ``IncompatiblePeer`` error. ``IncompatiblePeer`` wraps four distinct
        negotiation failures (host key / KEX / cipher / MAC), and only the
        first is addressable by ``enable_ssh_rsa_compat()``. Inspecting
        the four corresponding name-lists in the result tells you which
        list the server narrowed and to what.

        Pure-socket parsing (rather than driving ``paramiko.Transport``) is
        deliberate: the result reflects what the *server* advertises,
        independent of any process-global paramiko state mutated by
        ``enable_ssh_rsa_compat()`` or downstream code.

        Args:
            host: Hostname or IP address of the SFTP server.
            port: SSH port (default: 22).
            timeout: Socket timeout in seconds (default: 10).

        Returns:
            A dictionary with eleven entries:

            - ``"banner"`` -- the server's identification string (e.g.
                ``"SSH-2.0-OpenSSH_8.9p1"``).
            - The ten RFC 4253 § 7.1 name-lists, each as a Python ``list[str]``:
                ``kex_algorithms``, ``server_host_key_algorithms``,
                ``encryption_algorithms_ctos``, ``encryption_algorithms_stoc``,
                ``mac_algorithms_ctos``, ``mac_algorithms_stoc``,
                ``compression_algorithms_ctos``,
                ``compression_algorithms_stoc``, ``languages_ctos``,
                ``languages_stoc``.

        Raises:
            OSError: Socket-level failure (host unreachable, port refused,
                timeout, connection reset, unexpected EOF, or the server's
                first packet was not ``SSH_MSG_KEXINIT``).

        !!! example "Diagnose an `ssh-rsa`-only legacy server"

            ```python
            from remote_store.backends import SFTPUtils

            info = SFTPUtils.scan_host_algorithms("legacy.example.com")
            print(info["server_host_key_algorithms"])
            # ['ssh-rsa']  -> classic legacy server; on paramiko 5+,
            # call SFTPUtils.enable_ssh_rsa_compat() at process startup.
            ```

        !!! example "Diagnose a narrow-KEX server"

            ```python
            info = SFTPUtils.scan_host_algorithms("legacy.example.com")
            print(info["kex_algorithms"])
            # ['diffie-hellman-group14-sha1']  -> KEX narrowing;
            # widen via SFTPBackend(connect_kwargs={"disabled_algorithms": ...}).
            ```
        """
    @staticmethod
    def enable_ssh_rsa_compat() -> None:
        """Guarantee ``ssh-rsa`` (SHA-1) acceptance across paramiko's four host-key sites.

        Appends ``ssh-rsa`` to four paramiko class attributes if it is
        missing — future-proofing the consumer against eventual removal,
        and restoring state if downstream code has cleared it:

        1. ``paramiko.Transport._preferred_keys`` -- KEX host-key-algorithm
           negotiation.
        2. ``paramiko.Transport._key_info`` -- host-key parsing dispatch.
        3. ``paramiko.rsakey.RSAKey.HASHES`` -- signature-verification hash
           dispatch.
        4. ``paramiko.Transport._preferred_pubkeys`` -- client RSA public-key
           authentication signatures.

        Empirically verified across paramiko 2.12 / 3.0 / 3.5 / 4.0 / 5.0
        (see ``sdd/research/research-bk-198-paramiko-ssh-rsa-empirical.md``):

        - On paramiko **< 5.0** all four sites contain ``ssh-rsa`` by
          default, so a freshly-imported paramiko already negotiates against
          an ``ssh-rsa``-only server. The helper is a no-op (all four guards
          short-circuit) and is safe to call eagerly for forward
          compatibility.
        - On paramiko **>= 5.0** all four sites have ``ssh-rsa`` removed.
          Bare connect to an ``ssh-rsa``-only server fails immediately in
          ``Transport._parse_kex_init`` with ``IncompatiblePeer: no
          acceptable host key``. Calling this helper at process startup is
          the only way to restore the connection without monkey-patching
          ``connect_kwargs["disabled_algorithms"]`` on every connect.

        For KEX / cipher / MAC negotiation failures (e.g.
        ``IncompatiblePeer: no acceptable kex algorithm``), this helper is
        not the right tool — use ``connect_kwargs={"disabled_algorithms":
        ...}`` to widen those instead.

        ``disabled_algorithms`` cannot re-add a default-removed algorithm,
        so class-level patching is the only forward-compatible path. The
        patches are idempotent; calling this multiple times in the same
        process is safe.

        !!! warning "Process-global side effect"

            Every paramiko transport in this process will accept SHA-1 host
            keys for the lifetime of the process. Only call this if the
            consumer connects exclusively to servers under your operational
            control, or if you have explicitly evaluated the tradeoff for
            every server in the process.

            ``ssh-rsa`` is appended (not prepended) to the preferred lists, so
            modern algorithms are still negotiated first when the server
            offers them.

        !!! warning "Single-threaded startup only"

            The four read-then-write patches are not atomic; if two threads
            enter the helper at the same time they race on the rebind. Call
            once at process startup before any backend connect, not from a
            request-handling code path.

        !!! example

            ```python
            from remote_store.backends import SFTPUtils

            # Call once at process startup, before any SFTPBackend connect.
            SFTPUtils.enable_ssh_rsa_compat()
            ```
        """
class SFTPBackend(Backend):
    """SFTP backend using pure paramiko.

    ``move()`` attempts ``posix_rename`` (atomic on POSIX-compliant servers),
    then falls back to ``rename``, and finally to a stream copy followed by
    a delete.  Because atomicity cannot be guaranteed across all servers,
    ``ATOMIC_MOVE`` is not declared.

    Warning:
        **Not thread-safe for concurrent access.** This backend maintains a
        single SSH/SFTP connection (paramiko ``SFTPClient``), which is not
        safe to call from multiple threads simultaneously. Concurrent calls
        via ``SyncBackendAdapter`` and
        ``asyncio.gather`` will race on the shared socket and may hang or
        corrupt responses. Create one ``SFTPBackend`` instance per thread if
        you need parallel operations.

    Args:
        host: SFTP server hostname (required, non-empty).
        port: SSH port (default: 22).
        username: SSH username.
        password: SSH password.
        pkey: paramiko.PKey instance for key-based auth.
        base_path: Root path on the remote server (default: ``/``).
        host_key_policy: Host key verification policy
            (see ``SFTPUtils.HostKeyPolicy``). Accepts enum value or string.
        known_host_keys: Known hosts string (code-level override).
        host_keys_path: Path to known_hosts file (default: ``~/.ssh/known_hosts``).
        config: Optional config dict (may contain ``known_host_keys``).
        timeout: SSH connection timeout in seconds.
        connect_kwargs: Extra kwargs passed to ``SSHClient.connect()``.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _SFTP_CAPABILITIES
    def __init__(
        self,
        host: str,
        *,
        port: int = 22,
        username: str | None = None,
        password: str | Secret | None = None,
        pkey: Any = None,
        base_path: str = "/",
        host_key_policy: HostKeyPolicy | str = HostKeyPolicy.STRICT,
        known_host_keys: str | None = None,
        host_keys_path: str | None = None,
        config: dict[str, Any] | None = None,
        timeout: int = 10,
        connect_kwargs: dict[str, Any] | None = None,
        retry: RetryPolicy | None = None,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def check_health(self) -> None:
        """Confirm the SFTP connection works by ``stat``-ing the base path.

        Establishes the SSH/SFTP connection lazily if needed (retried at
        connection scope) and issues one ``stat`` round-trip.

        Raises:
            NotFound: If the configured base path does not exist.
            PermissionDenied: If the server denies access to the base path.
            BackendUnavailable: If the SSH/SFTP connection cannot be established.
        """
    def to_key(self, native_path: str) -> str:
    def native_path(self, path: str) -> str:
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with SFTP-specific details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="sftp"`` and ``details`` containing
            ``host``, ``port``, and ``base_path``.
        """
    def exists(self, path: str) -> bool:
        """Return ``True`` if a file or folder exists at *path*; never ``NotFound``.

        Issues one ``stat`` round-trip; a missing path returns ``False``.

        Raises:
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing regular file (``False`` if absent or a folder).

        Raises:
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if *path* is an existing directory (``False`` if absent or a file).

        Raises:
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def read(self, path: str) -> BinaryIO:
        """Open *path* for reading and return a buffered, streaming handle.

        Reads lazily over the SFTP channel (wrapped in a ``BufferedReader``), so
        memory stays constant regardless of file size; one extra ``stat``
        round-trip guards against *path* being a directory before the open.

        Raises:
            NotFound: If the file does not exist, or a path component is itself a
                file.
            InvalidPath: If *path* names a directory.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails mid-read.
        """
    def read_bytes(self, path: str) -> bytes:
        """Read and return the full file content as bytes.

        Prefetches and materialises the whole file in memory (unlike the lazy
        ``read`` stream).

        Raises:
            NotFound: If the file does not exist, or a path component is itself a
                file.
            InvalidPath: If *path* names a directory.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails mid-read.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path*, streaming it over the SFTP channel.

        The bytes are streamed straight to the destination file (no
        temp-and-rename), so a dropped connection mid-write can leave a partial
        or truncated file there — use ``write_atomic`` when readers must never
        see a half-written file. Missing parent directories are created first
        (one ``stat`` per ancestor).

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor of *path*
                exists as a regular file.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails mid-write.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Write *content* to *path* atomically via a temp file plus server rename.

        Readers never observe a partial file: the body is streamed to a hidden
        temp file in the destination directory, then promoted with
        ``posix_rename`` (atomic on POSIX-compliant servers). Servers without
        ``posix_rename`` fall back to a plain ``rename`` (non-atomic overwrite:
        the target is removed first), and the temp file is cleaned up on failure.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor of *path*
                exists as a regular file.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    @contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield a writable handle promoted to *path* atomically on clean exit.

        Writes stream to a hidden temp file in the destination directory; on
        clean exit it is promoted with ``posix_rename`` (atomic on POSIX
        servers, falling back to ``rename``), and on any exception the temp file
        is removed and *path* is left untouched.

        Raises:
            AlreadyExists: If the file exists and ``overwrite`` is ``False``.
            InvalidPath: If *path* names a directory, or an ancestor of *path*
                exists as a regular file.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the file at *path*.

        Raises:
            NotFound: If the file does not exist (or a path component is itself a
                file) and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a directory (use ``delete_folder``).
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete the folder at *path*.

        ``recursive=True`` walks and removes the subtree bottom-up (one round-trip
        per entry — not atomic; an interruption can leave the tree partially
        removed). ``recursive=False`` removes only an empty folder after checking
        it has no entries.

        Raises:
            NotFound: If the folder does not exist and ``missing_ok`` is ``False``.
            InvalidPath: If *path* names a file, not a folder.
            DirectoryNotEmpty: If the folder is non-empty and ``recursive`` is
                ``False``.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield files under *path*, one ``FileInfo`` at a time.

        Lazily walks the remote directory (``listdir_attr``); a missing *path*
        yields nothing. ``recursive`` descends via one directory-listing
        round-trip per folder (``max_depth`` bounds the descent). Failures other
        than a missing path surface as ``RemoteStoreError`` during iteration.
        """
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """Yield immediate subfolders of *path* as ``FolderEntry`` records.

        One directory-listing round-trip; a missing *path* yields nothing, and
        other failures surface as ``RemoteStoreError`` during iteration.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield the immediate files and folders under *path* in one listing.

        Overrides the base two-pass default with a single ``listdir_attr``
        round-trip, yielding ``FileInfo`` for files and ``FolderEntry`` for
        folders. A missing *path* yields nothing; other failures surface as
        ``RemoteStoreError`` during iteration.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return metadata for the file at *path* from a single ``stat`` round-trip.

        Raises:
            NotFound: If the file does not exist.
            InvalidPath: If *path* names a directory, not a file.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregate metadata for the folder at *path*.

        File count, total size, and latest modification time are gathered by
        recursively walking the whole subtree (one listing round-trip per
        folder), so cost scales with the number of descendants.

        Raises:
            NotFound: If the folder does not exist.
            InvalidPath: If *path* names a file, not a folder.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move or rename the file *src* to *dst*.

        Tries ``posix_rename`` first (atomic on POSIX-compliant servers), then a
        plain ``rename``, and finally a stream copy-then-delete. Because the
        outcome depends on server support, atomicity is not guaranteed across all
        servers and ``ATOMIC_MOVE`` is not declared. ``src == dst`` is a no-op;
        missing parent directories of *dst* are created first.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* names a directory, or an ancestor of
                *dst* exists as a regular file.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the file *src* to *dst* by streaming through the client.

        SFTP has no server-side copy, so the bytes round-trip through the client
        (download then upload); this is not atomic — an interruption can leave a
        partial file at *dst*. ``src == dst`` is a no-op; missing parent
        directories of *dst* are created first.

        Raises:
            NotFound: If *src* does not exist.
            InvalidPath: If *src* or *dst* names a directory, or an ancestor of
                *dst* exists as a regular file.
            AlreadyExists: If *dst* exists, ``src != dst``, and ``overwrite`` is
                ``False``.
            PermissionDenied: If the server denies access (``EACCES``).
            BackendUnavailable: If the SSH/SFTP connection cannot be established
                or fails.
        """
    def close(self) -> None:
    def unwrap(self, type_hint: type[T]) -> T:
    def __del__(self) -> None:
    def __repr__(self) -> str:
```

[52/69] src/remote_store/backends/_sqlalchemy.py (1508 rows, definitions)
---
```python
@runtime_checkable
class ResultSerializer(Protocol):
    """Converts SQL result rows to bytes in a specific format."""
    def serialize(self, rows: Sequence[Any], columns: Sequence[str], format: str) -> bytes:
        """Serialize rows with given column names to the specified format.

        Args:
            rows: Sequence of row tuples from SQL execution.
            columns: Column name list.
            format: Target format (``"parquet"``, ``"csv"``, ``"arrow"``).

        Returns:
            Serialized bytes.
        """
class ArrowSerializer:
    """Serializes SQL result sets via PyArrow.

    Converts rows + columns to a ``pyarrow.Table``, then writes to the
    requested format. Imports ``pyarrow`` lazily so that ``SQLBlobBackend``
    remains importable without it.
    """
    def serialize(self, rows: Sequence[Any], columns: Sequence[str], format: str) -> bytes:
        """Serialize rows to Parquet, CSV, or Arrow IPC."""
class SQLBlobBackend(_SQLAlchemyBaseBackend):
    """SQL key-value blob store implementing the full Backend contract.

    Uses a SQL table as key-value storage. Each row holds one "file"
    with its key, data, and metadata. SQLite receives WAL mode and
    PRAGMA tuning automatically.

    Supports all capabilities except ``LAZY_READ``.

    Every mutating operation runs inside a single database transaction, so
    ``write``, ``write_atomic``, ``move``, and ``copy`` are atomic — a failure
    rolls back with no partial row left behind (``ATOMIC_MOVE`` is advertised).

    Note:
        **Non-lazy reads and writes.** Both ``read()`` and ``write()``
        materialize the full content in memory. ``read()`` loads the
        entire BLOB before returning a stream (no ``LAZY_READ``).
        ``write()`` reads the full stream before issuing the SQL
        INSERT/UPDATE because BLOB columns require complete data in a
        single statement. For files larger than process memory, use a
        blob-storage backend (S3, Local, Azure) instead.

    Args:
        reject_write_under_file_ancestor: If ``True``, ``write`` /
            ``write_atomic`` / ``open_atomic`` / ``move`` / ``copy``
            issue one ``SELECT 1`` per slash-aligned ancestor of the
            target path and raise ``InvalidPath`` on the first
            regular-file hit, matching the cross-backend contract that
            hierarchical filesystems enforce natively. Default ``False``;
            paths without slashes short-circuit.
    """
    # Upper bound — runtime capabilities() may narrow for narrow-column schemas (create_table=False).
    CAPABILITIES: ClassVar[CapabilitySet] = _ALL_CAPABILITIES
    def __init__(
        self,
        url: str | None = None,
        *,
        engine: Engine | None = None,
        table_name: str = "remote_store_objects",
        create_table: bool = True,
        max_blob_size: int | None = None,
        reject_write_under_file_ancestor: bool = False,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with SQL blob details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="sql-blob"`` and ``details`` containing
            ``table_name``.
        """
    def exists(self, path: str) -> bool:
        """Return ``True`` if a key or key-prefix exists at *path*; never ``NotFound``.

        Folders are virtual — *path* counts as a folder when any key begins with
        ``path + "/"``. The root (``""``) always exists. Costs one or two
        ``SELECT``s.

        Raises:
            InvalidPath: If *path* is absolute or contains a ``..`` segment.
            BackendUnavailable: If the database is unreachable.
        """
    def is_file(self, path: str) -> bool:
        """Return ``True`` if an exact key exists at *path* (one ``SELECT``).

        Raises:
            InvalidPath: If *path* is absolute or contains a ``..`` segment.
            BackendUnavailable: If the database is unreachable.
        """
    def is_folder(self, path: str) -> bool:
        """Return ``True`` if any key begins with ``path + "/"`` (a virtual folder).

        The root is always a folder. Costs one ``SELECT``.

        Raises:
            InvalidPath: If *path* is absolute or contains a ``..`` segment.
            BackendUnavailable: If the database is unreachable.
        """
    def read(self, path: str) -> BinaryIO:
        """Return a binary stream over the stored BLOB for *path*.

        Loads the entire BLOB into memory before returning the stream — the read
        does not stream (``LAZY_READ`` is not advertised), so peak memory scales
        with the object size. For objects larger than process memory use a
        blob-storage backend (S3, Local, Azure).

        Raises:
            NotFound: If no key exists at *path*.
            InvalidPath: If *path* is empty, absolute, or malformed.
            BackendUnavailable: If the database is unreachable.
        """
    def read_bytes(self, path: str) -> bytes:
        """Return the full stored BLOB for *path* as bytes.

        Like ``read``, materialises the whole object in memory.

        Raises:
            NotFound: If no key exists at *path*.
            InvalidPath: If *path* is empty, absolute, or malformed.
            BackendUnavailable: If the database is unreachable.
        """
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Store *content* at *path* in a single, atomic transaction.

        The whole body is buffered in memory before the ``INSERT``/``UPDATE``
        (BLOB columns need the complete value in one statement — no streaming
        write), and the row is written inside one transaction, so a failure rolls
        back with no partial row left behind.

        Raises:
            AlreadyExists: If a key exists at *path* and ``overwrite`` is ``False``.
            InvalidPath: If *path* is empty/malformed, or (with the
                ``reject_write_under_file_ancestor`` opt-in) an ancestor key
                exists as a file.
            BackendUnavailable: If the database operation fails.
        """
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
        """Store *content* at *path* atomically (delegates to ``write``).

        SQL writes are already transactional, so this is exactly ``write``; the
        whole body is buffered first.

        Raises:
            AlreadyExists: If a key exists at *path* and ``overwrite`` is ``False``.
            InvalidPath: If *path* is empty/malformed, or (opt-in) an ancestor key
                exists as a file.
            BackendUnavailable: If the database operation fails.
        """
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
        """Yield an in-memory buffer committed to *path* atomically on clean exit.

        Writes accumulate in a ``BytesIO``; on exit the buffer is stored via
        ``write`` in one transaction. An exception before exit leaves *path*
        untouched.

        Raises:
            AlreadyExists: If a key exists at *path* and ``overwrite`` is ``False``.
            InvalidPath: If *path* is empty/malformed, or (opt-in) an ancestor key
                exists as a file.
            BackendUnavailable: If the database operation fails.
        """
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
        """Delete the row at *path* in one transaction.

        Raises:
            NotFound: If no key exists at *path* and ``missing_ok`` is ``False``.
            InvalidPath: If *path* is empty, absolute, or malformed.
            BackendUnavailable: If the database operation fails.
        """
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
        """Delete every key under the virtual folder *path*.

        Folders are key prefixes, not stored rows, so a folder "exists" only when
        it has children: a non-recursive call on an existing folder therefore
        always raises ``DirectoryNotEmpty``. ``recursive=True`` deletes all keys
        under ``path + "/"`` in one atomic transaction.

        Raises:
            NotFound: If no key exists under *path* and ``missing_ok`` is ``False``.
            DirectoryNotEmpty: If ``recursive`` is ``False`` (an existing virtual
                folder is never empty).
            InvalidPath: If *path* is empty, absolute, or malformed.
            BackendUnavailable: If the database operation fails.
        """
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
        """Yield files under *path*.

        One ``SELECT`` fetches every key under the prefix; folder structure is
        derived from ``/`` in the key suffix, and ``recursive`` / ``max_depth``
        filter client-side. A missing prefix yields nothing.

        Raises:
            BackendUnavailable: If the database operation fails, surfaced during
                iteration.
        """
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
        """Yield immediate virtual subfolders of *path* as ``FolderEntry`` records.

        One ``SELECT`` over keys under the prefix; folder names are the distinct
        first segments of the key suffixes. A missing prefix yields nothing.

        Raises:
            BackendUnavailable: If the database operation fails, surfaced during
                iteration.
        """
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
        """Yield the immediate files and virtual folders under *path* in one ``SELECT``.

        Overrides the base two-pass default: a single query over the prefix
        yields ``FileInfo`` for direct-child keys and ``FolderEntry`` for the
        distinct first suffix segments. A missing prefix yields nothing.

        Raises:
            BackendUnavailable: If the database operation fails, surfaced during
                iteration.
        """
    def get_file_info(self, path: str) -> FileInfo:
        """Return metadata for the file at *path* from one ``SELECT``.

        Raises:
            NotFound: If no key exists at *path*.
            InvalidPath: If *path* is empty, absolute, or malformed.
            BackendUnavailable: If the database is unreachable.
        """
    def get_folder_info(self, path: str) -> FolderInfo:
        """Return aggregate metadata for the virtual folder *path*.

        File count, total size, and latest modification time come from one
        aggregate ``SELECT`` (``COUNT``/``SUM``/``MAX``) over keys under the
        prefix — no per-file round-trips.

        Raises:
            NotFound: If no key exists under *path*.
            InvalidPath: If *path* is absolute or malformed.
            BackendUnavailable: If the database is unreachable.
        """
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Move (rename) the key *src* to *dst* in one atomic transaction.

        Implemented as an ``UPDATE`` of the row's key — the BLOB is never
        transferred through Python. The whole operation (source check, optional
        destination replace, rename) runs in one transaction, so a failure rolls
        back cleanly. ``src == dst`` verifies the source exists and is otherwise
        a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists and ``overwrite`` is ``False``.
            InvalidPath: If *src* or *dst* is malformed, or (opt-in) an ancestor
                of *dst* exists as a file.
            BackendUnavailable: If the database operation fails.
        """
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
        """Copy the key *src* to *dst* in one atomic transaction.

        Implemented as a single ``INSERT ... SELECT``, so the BLOB is duplicated
        entirely inside the database — no bytes pass through Python. ``src == dst``
        verifies the source exists and is otherwise a no-op.

        Raises:
            NotFound: If *src* does not exist.
            AlreadyExists: If *dst* exists and ``overwrite`` is ``False``.
            InvalidPath: If *src* or *dst* is malformed, or (opt-in) an ancestor
                of *dst* exists as a file.
            BackendUnavailable: If the database operation fails.
        """
    def glob(self, pattern: str) -> Iterator[FileInfo]:
        """Yield files whose key matches the glob *pattern*.

        Narrows SQL-side with a prefix ``LIKE`` where the pattern allows (on
        every dialect — SQLite's native ``GLOB`` is deliberately avoided because
        it mishandles ``**``), then applies the full glob regex to each row.
        Costs one ``SELECT``.

        Raises:
            BackendUnavailable: If the database operation fails, surfaced during
                iteration.
        """
    def __repr__(self) -> str:
class SQLQueryBackend(_SQLAlchemyBaseBackend):
    """Read-only SQL query materializer implementing a subset of the Backend contract.

    Maps path keys to SQL queries. On ``read()``, executes the query and
    serializes the result set to the format implied by the key's file
    extension (Parquet, CSV, or Arrow IPC).

    Capabilities: ``READ``, ``LIST``, ``METADATA``, ``GLOB``,
    ``SEEKABLE_READ``.
    """
    CAPABILITIES: ClassVar[CapabilitySet] = _QUERY_CAPABILITIES
    def __init__(
        self,
        url: str | None = None,
        *,
        engine: Engine | None = None,
        queries: dict[str, str] | None = None,
        strict: bool = True,
        serializer: ResultSerializer | None = None,
    ) -> None:
    @property
    def name(self) -> str:
    @property
    def capabilities(self) -> CapabilitySet:
    def resolve(self, path: str) -> ResolutionPlan:
        """Return a ``ResolutionPlan`` with SQL query details.

        Args:
            path: Backend-relative key.

        Returns:
            Plan with ``kind="sql-query"`` and ``details`` containing
            ``source`` and ``format``.
        """
    def exists(self, path: str) -> bool:
    def is_file(self, path: str) -> bool:
    def is_folder(self, path: str) -> bool:
    def read(self, path: str) -> BinaryIO:
    def read_bytes(self, path: str) -> bytes:
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def open_atomic(self, path: str, *, overwrite: bool = False) -> contextlib.AbstractContextManager[BinaryIO]:
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
    def list_folders(self, path: str) -> Iterator[FolderEntry]:
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
    def get_file_info(self, path: str) -> FileInfo:
    def get_folder_info(self, path: str) -> FolderInfo:
    def glob(self, pattern: str) -> Iterator[FileInfo]:
    def __repr__(self) -> str:
```

[53/69] src/remote_store/ext/__init__.py (20 rows, definitions)

[54/69] src/remote_store/ext/arrow.py (429 rows, definitions)
---
```python
class StoreFileSystemHandler(pafs.FileSystemHandler):  # type: ignore[misc]
    """``pyarrow.fs.FileSystemHandler`` backed by a ``Store``.

    Args:
        store: The Store to expose as a PyArrow filesystem.
        materialization_threshold: Max file size (bytes) for Tier 2 full-file
            materialization in ``open_input_file``. Default 64 MB.
        write_spill_threshold: Max in-memory buffer size (bytes) for
            ``_StoreSink`` before spilling to disk. Default 64 MB.

    **Thread safety:** The handler itself holds no shared mutable state. PyArrow's
    C++ layer may call handler methods from background threads (with the GIL
    acquired). Thread safety therefore depends on the backend: ``MemoryBackend``
    uses a lock (safe), ``LocalBackend`` relies on OS file semantics (safe),
    cloud backends use thread-safe HTTP clients (safe). If using a custom backend,
    ensure its methods are safe under concurrent calls.
    """
    def __init__(
        self,
        store: Store,
        materialization_threshold: int = 64 * 1024 * 1024,
        write_spill_threshold: int = 64 * 1024 * 1024,
    ) -> None:
    def __eq__(self, other: object) -> bool:
    def __ne__(self, other: object) -> bool:
    # region: identity (PA-003, PA-006)
    def get_type_name(self) -> str:
    @staticmethod
    def normalize_path(path: str) -> str:
    # region: file info (PA-007, PA-008)
    def get_file_info(self, paths: list[str]) -> list[pafs.FileInfo]:
    def get_file_info_selector(self, selector: Any) -> list[pafs.FileInfo]:  # noqa: ANN401
    # region: read operations (PA-009, PA-010)
    def open_input_stream(self, path: str) -> Any:  # noqa: ANN401
    def open_input_file(self, path: str) -> Any:  # noqa: ANN401
    # region: write operations (PA-011, PA-012)
    def open_output_stream(self, path: str, metadata: Any = None) -> Any:  # noqa: ANN401
    def open_append_stream(self, path: str, metadata: Any = None) -> Any:  # noqa: ANN401
    # region: delete and directory operations (PA-013 through PA-015)
    def delete_file(self, path: str) -> None:
    def create_dir(self, path: str, recursive: bool = True) -> None:
    def delete_dir(self, path: str) -> None:
    def delete_dir_contents(self, path: str, *, missing_dir_ok: bool = False) -> None:
    def delete_root_dir_contents(self) -> None:
    # region: move and copy (PA-017, PA-018)
    def move(self, src: str, dest: str) -> None:
    def copy_file(self, src: str, dest: str) -> None:
def pyarrow_fs(
    store: Store,
    *,
    materialization_threshold: int = 64 * 1024 * 1024,
    write_spill_threshold: int = 64 * 1024 * 1024,
) -> pafs.PyFileSystem:
    """Create a ``pyarrow.fs.PyFileSystem`` backed by *store*.

    Args:
        store: The Store to expose.
        materialization_threshold: See ``StoreFileSystemHandler``.
        write_spill_threshold: See ``StoreFileSystemHandler``.
    """
```

[55/69] src/remote_store/ext/batch.py (298 rows, definitions)
---
```python
@dataclasses.dataclass(frozen=True)
class BatchResult:
    """Outcome of a batch operation.

    Attributes:
        succeeded: Paths that completed without error.
        failed: Mapping from path to the error that occurred.
    """
    succeeded: tuple[str, ...]
    failed: dict[str, RemoteStoreError]
    @property
    def all_succeeded(self) -> bool:
        """``True`` when every path succeeded."""
    @property
    def total(self) -> int:
        """Total number of paths processed (succeeded + failed)."""
def batch_delete(
    store: Store,
    paths: Iterable[str],
    *,
    missing_ok: bool = False,
    stop_on_error: bool = False,
    concurrent: bool = False,
    max_workers: int | None = None,
) -> BatchResult:
    """Delete multiple files, collecting errors.

    Args:
        store: The Store to delete from.
        paths: File paths to delete.
        missing_ok: Forwarded to each ``store.delete()`` call.
        stop_on_error: Stop on first ``RemoteStoreError`` (sequential only).
        concurrent: Use a thread pool for parallel execution.
        max_workers: Max threads (forwarded to ``ThreadPoolExecutor``).

    Returns:
        A ``BatchResult`` with succeeded/failed paths.

    Raises:
        ValueError: If both ``concurrent`` and ``stop_on_error`` are True.
    """
def batch_copy(
    store: Store,
    pairs: Iterable[tuple[str, str]],
    *,
    overwrite: bool = False,
    stop_on_error: bool = False,
    concurrent: bool = False,
    max_workers: int | None = None,
) -> BatchResult:
    """Copy multiple files, collecting errors.

    Args:
        store: The Store to copy within.
        pairs: ``(src, dst)`` tuples.
        overwrite: Forwarded to each ``store.copy()`` call.
        stop_on_error: Stop on first ``RemoteStoreError`` (sequential only).
        concurrent: Use a thread pool for parallel execution.
        max_workers: Max threads (forwarded to ``ThreadPoolExecutor``).

    Returns:
        A ``BatchResult`` with succeeded/failed source paths.

    Raises:
        ValueError: If both ``concurrent`` and ``stop_on_error`` are True.
    """
def batch_exists(
    store: Store,
    paths: Iterable[str],
    *,
    concurrent: bool = False,
    max_workers: int | None = None,
) -> dict[str, bool]:
    """Check existence of multiple paths.

    Unlike ``batch_delete()`` and ``batch_copy()``, this function does
    **not** catch errors — any exception from ``store.exists()`` propagates
    immediately.

    Args:
        store: The Store to query.
        paths: Paths to check.
        concurrent: Use a thread pool for parallel execution.
        max_workers: Max threads (forwarded to ``ThreadPoolExecutor``).

    Returns:
        Dict mapping each path to ``True``/``False``.
    """
```

[56/69] src/remote_store/ext/cache.py (640 rows, definitions)
---
```python
@dataclasses.dataclass(frozen=True)
class CacheStats:
    """Snapshot of cache hit/miss statistics."""
    hits: int
    misses: int
    size: int
@runtime_checkable
class CacheBackend(Protocol):
    """Protocol for pluggable cache backends.

    Implement this protocol to provide a custom cache backend to
    ``cache(store, cache_backend=my_backend)``. The default implementation
    is ``MemoryCache``.
    """
    def get(self, key: tuple[str, ...]) -> Any:
        """Return the cached value, or raise ``KeyError`` on a cache miss."""
    def set(self, key: tuple[str, ...], value: Any, ttl: float) -> None:
        """Store *value* under *key* with a time-to-live in seconds."""
    def delete(self, key: tuple[str, ...]) -> None:
        """Remove *key* from the cache (no-op if absent)."""
    def clear(self) -> None:
        """Remove all entries from the cache."""
    def clear_prefix(self, prefix: str) -> None:
        """Remove all entries whose first key component matches *prefix*."""
    def size(self) -> int:
        """Return the number of entries currently in the cache."""
class MemoryCache:
    """Thread-safe in-memory cache backend with lazy TTL eviction.

    Entries are stored as ``{key: (value, expiry)}`` where *expiry* is a
    ``time.monotonic()`` deadline.

    When *max_entries* is set, the cache evicts the least-recently-used
    entry when the limit is exceeded (LRU eviction).  Without a bound,
    metadata entries (``exists``, ``is_file``, listings) for many distinct
    paths can grow without limit during the TTL window.
    """
    def __init__(self, *, max_entries: int | None = None) -> None:
    def get(self, key: tuple[str, ...]) -> Any:
        """Return cached value or raise ``KeyError`` on miss/expiry."""
    def set(self, key: tuple[str, ...], value: Any, ttl: float) -> None:
        """Store *value* with a TTL in seconds."""
    def delete(self, key: tuple[str, ...]) -> None:
        """Remove a single entry (no-op if absent)."""
    def clear(self) -> None:
        """Remove all entries."""
    def clear_prefix(self, prefix: str) -> None:
        """Remove all entries whose first key element equals *prefix*."""
    def clear_prefixes(self, prefixes: frozenset[str]) -> None:
        """Remove all entries whose first key element is in *prefixes*.

        Single dict rebuild instead of one per prefix — O(n) vs O(k*n).
        """
    def size(self) -> int:
        """Return count of non-expired entries."""
class CachedStore(ProxyStore):
    """Proxy Store that caches read operations with TTL-based expiration.

    All ``Store`` methods are delegated to the inner store. Read-only
    methods use the cache; mutating methods invalidate affected entries.
    Only methods with additional behavior (``invalidate``, ``clear_cache``,
    ``ping``, ``close``, ``child``) are documented individually below.

    Do not construct directly -- use ``cache()``.
    """
    def __init__(
        self,
        inner: Store,
        *,
        ttl: float,
        max_content_size: int | None,
        max_listing_size: int | None,
        max_entries: int | None,
        cache_backend: CacheBackend | None,
        _prefix: str = "",
    ) -> None:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    @property
    def stats(self) -> CacheStats:
        """Snapshot of cache hit/miss statistics."""
    def invalidate(self, path: str) -> None:
        """Remove all cached entries for *path* and its ancestor directories."""
    def clear_cache(self) -> None:
        """Remove all cached entries."""
    def __repr__(self) -> str:
    def exists(self, path: str) -> bool:
    def is_file(self, path: str) -> bool:
    def is_folder(self, path: str) -> bool:
    def read_bytes(self, path: str) -> bytes:
    def get_file_info(self, path: str) -> FileInfo:
    def get_folder_info(self, path: str, *, max_depth: int | None = None) -> FolderInfo:
    def head(self, path: str) -> WriteResult:
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
    def list_folders(
        self, path: str, *, pattern: str | None = None, max_depth: int | None = None
    ) -> Iterator[FolderEntry]:
    def glob(self, pattern: str) -> Iterator[FileInfo]:
    def read_text(self, path: str, *, encoding: str = "utf-8", errors: str = "strict") -> str:
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_text(
        self,
        path: str,
        text: str,
        *,
        encoding: str = "utf-8",
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
def cache(
    store: Store,
    *,
    ttl: float = 300.0,
    max_content_size: int | None = None,
    max_listing_size: int | None = None,
    max_entries: int | None = None,
    cache_backend: CacheBackend | None = None,
) -> CachedStore:
    """Wrap a Store with read-through caching.

    Args:
        store: The Store to wrap.
        ttl: Time-to-live in seconds for cache entries (default 300).
        max_content_size: Maximum byte length for ``read_bytes`` caching.
            Files larger than this are returned without caching. ``None`` means
            unlimited.
        max_listing_size: Maximum number of items in a listing result
            (``iter_children``, ``list_files``, ``list_folders``, ``glob``)
            for caching. Listings with more items than this are returned
            without caching. ``None`` means unlimited.
        max_entries: Maximum number of cache entries. When exceeded, the
            least-recently-used entry is evicted. ``None`` means no limit.
            Ignored when *cache_backend* is provided.
        cache_backend: Optional custom cache. When ``None``, a
            ``MemoryCache`` is created.

    Returns:
        A ``CachedStore`` proxy.

    Raises:
        ValueError: If *ttl*, *max_content_size*, or *max_listing_size*
            is not positive when set.
    """
```

[57/69] src/remote_store/ext/dagster.py (805 rows, definitions)
---
```python
@runtime_checkable
class Serializer(Protocol):
    """Protocol for pluggable serializers.

    Implement this to provide a custom serializer to
    ``dagster_io_manager(store, serializer=my_serializer)``.
    """
    extension: str
    def serialize(self, obj: Any) -> bytes:
        """Convert a Python object to bytes."""
    def deserialize(self, data: bytes) -> Any:
        """Convert bytes back to a Python object."""
class PickleSerializer:
    """Pickle-based serializer. Universal; opaque format."""
    extension: str = ".pkl"
    def serialize(self, obj: Any) -> bytes:
        """Serialize using pickle."""
    def deserialize(self, data: bytes) -> Any:
        """Deserialize using pickle."""
class JsonSerializer:
    """JSON serializer. JSON-serializable objects only."""
    extension: str = ".json"
    def serialize(self, obj: Any) -> bytes:
        """Serialize to JSON bytes."""
    def deserialize(self, data: bytes) -> Any:
        """Deserialize from JSON bytes."""
class ParquetSerializer:
    """Parquet serializer via PyArrow. DataFrames and Arrow Tables."""
    extension: str = ".parquet"
    def __init__(self) -> None:
    def serialize(self, obj: Any) -> bytes:
        """Serialize a DataFrame to Parquet bytes."""
    def deserialize(self, data: bytes) -> pa.Table:
        """Deserialize Parquet bytes to a PyArrow Table."""
def dagster_io_manager(
    store: Store,
    *,
    serializer: str | Serializer = "pickle",
) -> IOManager:  # type: ignore[type-arg]
    """Wrap a Store as a Dagster IOManager.

    Args:
        store: An existing Store instance. The caller owns its lifecycle —
            the IO manager does not close the Store.
        serializer: ``"pickle"`` (default), ``"json"``, ``"parquet"``, or a
            custom object satisfying the ``Serializer`` protocol.

    Returns:
        A Dagster ``IOManager`` backed by the given Store.

    Raises:
        ValueError: If *serializer* is an unrecognized string.
    """
def dagster_dataset_io_manager(store: Store) -> IOManager:  # type: ignore[type-arg]
    """Wrap a Store as a Dagster IOManager using ParquetDatasetStore.

    Unlike ``dagster_io_manager`` which serializes objects to single files,
    this manager writes Parquet datasets (multi-file with manifest) via
    ``ParquetDatasetStore``.

    Args:
        store: An existing Store instance. The caller owns its lifecycle.

    Returns:
        A Dagster ``IOManager`` backed by ParquetDatasetStore.
    """
class DagsterStoreResource(ConfigurableResource):  # type: ignore[misc,type-arg]
    """Dagster resource that constructs a Store from config fields.

    Unlike stateless utility extensions, this class owns Store lifecycle
    because it is a Dagster Resource with ``setup_for_execution`` /
    ``teardown_after_execution`` hooks.

    Attributes:
        backend_type: Backend type string (e.g. ``"local"``, ``"s3"``, ``"memory"``).
            Must be registered in the backend factory registry.
        backend_options: Keyword arguments passed to the backend constructor.
        root_path: Optional root path for the Store.
    """
    backend_type: str
    backend_options: dict[str, Any] = {}
    root_path: str = ""
    def setup_for_execution(self, context: InitResourceContext) -> None:
        """Instantiate and cache the Store (called by Dagster before execution).

        Raises:
            ValueError: If *backend_type* is not registered, or if the backend
                constructor rejects the supplied *backend_options*.
        """
    def teardown_after_execution(self, context: InitResourceContext) -> None:
        """Close the Store and release resources (called by Dagster after execution)."""
    def get_store(self) -> Store:
        """Return the underlying Store instance.

        Returns:
            The Store constructed during ``setup_for_execution``.

        Raises:
            RuntimeError: If called before ``setup_for_execution`` has run.
        """
class RemoteStoreIOManager(ConfigurableIOManagerFactory):  # type: ignore[misc,type-arg]
    """IO manager factory that constructs a Store from config fields.

    Embeds backend configuration directly so the IO manager owns the full
    Store lifecycle (setup and teardown). For direct Store access in assets,
    use ``DagsterStoreResource`` as a separate resource.

    Attributes:
        backend_type: Backend type string (e.g. ``"local"``, ``"s3"``, ``"memory"``).
        backend_options: Keyword arguments passed to the backend constructor.
        root_path: Optional root path for the Store.
        serializer: Serializer name. Use ``"parquet-dataset"`` for multi-file
            Parquet dataset output via ``ParquetDatasetStore``.
    """
    backend_type: str
    backend_options: dict[str, Any] = {}
    root_path: str = ""
    serializer: str = "pickle"
    def setup_for_execution(self, context: InitResourceContext) -> None:
        """Build and cache the Store before execution."""
    def teardown_after_execution(self, context: InitResourceContext) -> None:
        """Close the Store and release resources."""
    def create_io_manager(self, context: Any) -> IOManager:  # type: ignore[type-arg]
        """Construct the IOManager for the given execution context.

        Returns:
            A ``_DatasetIOManagerImpl`` when ``serializer="parquet-dataset"``,
            otherwise a ``_RemoteStoreIOManagerImpl`` with the resolved serializer.

        Raises:
            RuntimeError: If called before ``setup_for_execution``.
            ValueError: If *serializer* is an unrecognized string.
        """
class RemoteStoreComputeLogManager(  # type: ignore[misc]
    TruncatingCloudStorageComputeLogManager,
    ConfigurableClass,
):
    """Captures op/step ``stdout`` / ``stderr`` to any remote-store backend.

    A Dagster ``ComputeLogManager`` wired into ``dagster.yaml`` as an instance
    component. Logs are captured to a local staging directory at the
    file-descriptor level, then uploaded to a ``Store`` the manager builds
    itself from ``backend_type`` + ``backend_options``. When ``upload_interval``
    is set, partial uploads also run periodically while a step executes so the
    Dagster UI can tail them. Subclasses Dagster's
    ``TruncatingCloudStorageComputeLogManager`` (capture-then-upload machinery,
    50 MB upload truncation) and ``ConfigurableClass`` (``dagster.yaml``
    plumbing).

    Configure it in ``dagster.yaml``:

    ```yaml
    compute_logs:
      module: remote_store.ext.dagster
      class: RemoteStoreComputeLogManager
      config:
        backend_type: s3
        backend_options:
          bucket: my-logs-bucket
        root_path: dagster/compute-logs
        upload_interval: 30
    ```

    Attributes:
        backend_type: Registered backend type (``"local"``, ``"s3"``, ``"sftp"``,
            ``"azure"``, ``"memory"``, ...).
        backend_options: Keyword arguments for the backend constructor.
        root_path: Store root prefix applied to every log object.
        local_dir: Local staging directory for capture. Defaults to the
            system temp directory.
        prefix: Path prefix within the Store. Defaults to ``"dagster"``.
        skip_empty_files: Skip uploading zero-byte log files.
        upload_interval: Seconds between partial uploads while a step runs;
            ``None`` (default) disables live tailing.
    """
    def __init__(
        self,
        backend_type: str,
        backend_options: dict[str, Any] | None = None,
        root_path: str = "",
        local_dir: str | None = None,
        prefix: str = "dagster",
        skip_empty_files: bool = False,
        upload_interval: int | None = None,
        inst_data: ConfigurableClassData | None = None,
    ) -> None:
        """Build the Store, validate its capabilities, and wire the local manager.

        Raises:
            ValueError: If *backend_type* is not registered, if the backend
                rejects *backend_options*, or if the backend is missing a
                capability the manager requires (``READ``, ``WRITE``,
                ``DELETE``, ``METADATA``, ``LIST``).
        """
    @property
    def inst_data(self) -> ConfigurableClassData | None:
        """The ``ConfigurableClassData`` this manager was rehydrated from, if any."""
    @classmethod
    def config_type(cls) -> dict[str, Any]:
        """The ``dagster.yaml`` config schema for this manager."""
    @classmethod
    def from_config_value(
        cls, inst_data: ConfigurableClassData | None, config_value: Mapping[str, Any]
    ) -> RemoteStoreComputeLogManager:
        """Construct a manager from a validated ``dagster.yaml`` config value."""
    @property
    def local_manager(self) -> LocalComputeLogManager:
        """The ``LocalComputeLogManager`` that stages captures before upload."""
    @property
    def upload_interval(self) -> int | None:
        """Seconds between partial uploads, or ``None`` when live tailing is off."""
    def download_from_cloud_storage(
        self, log_key: Sequence[str], io_type: ComputeIOType, partial: bool = False
    ) -> None:
        """Stream a log object from the Store into the local staging file."""
    def cloud_storage_has_logs(self, log_key: Sequence[str], io_type: ComputeIOType, partial: bool = False) -> bool:
        """Return whether the Store holds a log object for this key."""
    def display_path_for_type(  # type: ignore[override]  # base hint is `str`; `None` until capture completes (matches S3ComputeLogManager)
        self, log_key: Sequence[str], io_type: ComputeIOType
    ) -> str | None:
        """A human-readable Store location for the Dagster UI, once capture is done."""
    def download_url_for_type(self, log_key: Sequence[str], io_type: ComputeIOType) -> str | None:
        """No signed-URL primitive in v1 — the webserver streams logs itself."""
    def delete_logs(self, log_key: Sequence[str] | None = None, prefix: Sequence[str] | None = None) -> None:
        """Delete captured logs by ``log_key`` or by ``prefix``, local and remote.

        Raises:
            CheckError: If neither *log_key* nor *prefix* is given.
        """
    def get_log_keys_for_log_key_prefix(
        self, log_key_prefix: Sequence[str], io_type: ComputeIOType
    ) -> Sequence[Sequence[str]]:
        """Enumerate the stored log keys under a log-key prefix."""
    def on_subscribe(self, subscription: CapturedLogSubscription) -> None:
        """Register a UI live-tail subscription with the polling manager."""
    def on_unsubscribe(self, subscription: CapturedLogSubscription) -> None:
        """Deregister a UI live-tail subscription from the polling manager."""
    def dispose(self) -> None:
        """Dispose the subscription and local managers and close the Store."""
```

[58/69] src/remote_store/ext/glob.py (69 rows, definitions)
---
```python
def glob_files(store: Store, pattern: str) -> Iterator[FileInfo]:
    """Match files against a glob pattern.

    When the backend supports ``Capability.GLOB``, delegates to
    ``store.glob()`` for native pattern matching. Otherwise extracts the
    longest non-wildcard prefix, lists files under that prefix, and
    filters client-side.

    Args:
        store: The Store to search.
        pattern: Glob pattern relative to the store root
            (e.g., ``"data/*.csv"``, ``"**/*.txt"``).

    Returns:
        Iterator of matching ``FileInfo`` objects.
    """
```

[59/69] src/remote_store/ext/integrity.py (131 rows, definitions)
---
```python
def checksum(store: Store, path: str, algorithm: str = "sha256") -> tuple[str, str]:
    """Compute the checksum of a file in the store.

    Reads the file in chunks (never fully materialized in memory) and
    returns a ``(algorithm, hex_digest)`` tuple.

    Args:
        store: The Store to read from.
        path: Store-relative file path.
        algorithm: Hash algorithm name (default ``"sha256"``).

    Returns:
        Tuple of ``(algorithm, hex_digest)`` with lowercase values.

    Raises:
        NotFound: If the file does not exist.
        ValueError: If the algorithm is not supported by ``hashlib``.
    """
def content_digest(store: Store, path: str, algorithm: str = "sha256") -> ContentDigest:
    """Compute the content digest of a file in the store.

    Like [checksum][remote_store.ext.integrity.checksum], but returns a
    [ContentDigest][remote_store._models.ContentDigest] instead of a raw tuple.

    Args:
        store: The Store to read from.
        path: Store-relative file path.
        algorithm: Hash algorithm name (default ``"sha256"``).

    Returns:
        A [ContentDigest][remote_store._models.ContentDigest] with normalized algorithm and hex value.

    Raises:
        NotFound: If the file does not exist.
        ValueError: If the algorithm is not supported by ``hashlib``.
    """
def verify(
    store: Store,
    path: str,
    expected: str,
    algorithm: str = "sha256",
) -> bool:
    """Verify a file's checksum against an expected hex value.

    Args:
        store: The Store to read from.
        path: Store-relative file path.
        expected: Expected hex digest (case-insensitive).
        algorithm: Hash algorithm name (default ``"sha256"``).

    Returns:
        ``True`` if the computed digest matches *expected*.

    Raises:
        NotFound: If the file does not exist.
        ValueError: If the algorithm is not supported by ``hashlib``.
    """
def verify_hex(
    store: Store,
    path: str,
    algorithm: str,
    expected_hex: str,
) -> bool:
    """Verify a file's checksum given an algorithm and expected hex value.

    Args:
        store: The Store to read from.
        path: Store-relative file path.
        algorithm: Hash algorithm name.
        expected_hex: Expected hex digest (case-insensitive).

    Returns:
        ``True`` if the computed digest matches *expected_hex*.

    Raises:
        NotFound: If the file does not exist.
        ValueError: If the algorithm is not supported by ``hashlib``.
    """
```

[60/69] src/remote_store/ext/observe.py (561 rows, definitions)
---
```python
def set_correlation_id(cid: str | None) -> Token[str | None]:
    """Set the correlation ID for the current context.

    Returns a token that can be used to reset the value:

    ```python
    token = set_correlation_id("req-123")
    # ... operations here will have correlation_id="req-123" ...
    _correlation_id.reset(token)
    ```

    Args:
        cid: Correlation ID string, or ``None`` to clear.

    Returns:
        A ``Token`` for resetting the value.
    """
@dataclasses.dataclass(frozen=True)
class StoreEvent:
    """Immutable record of a single Store operation."""
    operation: str
    path: str
    backend: str
    started_at: float
    duration_ms: float
    error: Exception | None
    metadata: dict[str, Any]
    correlation_id: str | None
class ObservedStore(ProxyStore):
    """Proxy Store that fires observation hooks on every public method.

    All ``Store`` methods are delegated to the inner store. Only methods
    with additional behavior (``ping``, ``close``, ``child``) are
    documented individually below; the remaining overrides add hook
    dispatch and are otherwise transparent.

    Do not construct directly -- use ``observe()``.
    """
    def __init__(
        self,
        inner: Store,
        *,
        hooks: dict[str, Any],
        around: Any | None,
    ) -> None:
    def __eq__(self, other: object) -> bool:
    def __hash__(self) -> int:
    def __repr__(self) -> str:
    # region: public method overrides
    def ping(self) -> None:  # noqa: D401
        """Delegate ping to inner store."""
    def close(self) -> None:  # noqa: D401
        """Delegate close to inner store."""
    def child(self, subpath: str) -> Store:
        """Return an observed child store."""
    def to_key(self, path: str) -> str:
    def unwrap(self, type_hint: type[T]) -> T:
    def native_path(self, key: str) -> str:
    def resolve(self, key: str) -> ResolutionPlan:
        """Resolve a key to a ``ResolutionPlan``, emitting observation events.

        Event metadata is intentionally empty to prevent sensitive
        ``details`` from flowing through observation callbacks.
        """
    def supports(self, capability: Capability) -> bool:
    def exists(self, path: str) -> bool:
    def is_file(self, path: str) -> bool:
    def is_folder(self, path: str) -> bool:
    def read(self, path: str) -> BinaryIO:
    def read_bytes(self, path: str) -> bytes:
    def read_seekable(self, path: str) -> BinaryIO:
    def read_text(self, path: str, *, encoding: str = "utf-8", errors: str = "strict") -> str:
    def write(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_text(
        self,
        path: str,
        text: str,
        *,
        encoding: str = "utf-8",
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    def write_atomic(
        self,
        path: str,
        content: WritableContent,
        *,
        overwrite: bool = False,
        metadata: Mapping[str, str] | None = None,
    ) -> WriteResult:
    @contextlib.contextmanager
    def open_atomic(self, path: str, *, overwrite: bool = False) -> Iterator[BinaryIO]:
    def delete(self, path: str, *, missing_ok: bool = False) -> None:
    def delete_folder(self, path: str, *, recursive: bool = False, missing_ok: bool = False) -> None:
    def iter_children(self, path: str) -> Iterator[FileInfo | FolderEntry]:
    def list_files(
        self,
        path: str,
        *,
        recursive: bool = False,
        pattern: str | None = None,
        max_depth: int | None = None,
    ) -> Iterator[FileInfo]:
    def glob(self, pattern: str) -> Iterator[FileInfo]:
    def list_folders(
        self, path: str, *, pattern: str | None = None, max_depth: int | None = None
    ) -> Iterator[FolderEntry]:
    def get_file_info(self, path: str) -> FileInfo:
    def get_folder_info(self, path: str, *, max_depth: int | None = None) -> FolderInfo:
    def head(self, path: str) -> WriteResult:
    def move(self, src: str, dst: str, *, overwrite: bool = False) -> None:
    def copy(self, src: str, dst: str, *, overwrite: bool = False) -> None:
def observe(
    store: Store,
    *,
    on_read: Callable[[StoreEvent], None] | None = None,
    on_write: Callable[[StoreEvent], None] | None = None,
    on_delete: Callable[[StoreEvent], None] | None = None,
    on_copy: Callable[[StoreEvent], None] | None = None,
    on_move: Callable[[StoreEvent], None] | None = None,
    on_list: Callable[[StoreEvent], None] | None = None,
    on_ping: Callable[[StoreEvent], None] | None = None,
    on_error: Callable[[StoreEvent], None] | None = None,
    on_any: Callable[[StoreEvent], None] | None = None,
    around: Callable[[str, str, str], AbstractContextManager[None]] | None = None,
) -> ObservedStore:
    """Wrap a Store with observation hooks.

    Args:
        store: The Store to observe.
        on_read: Fires after read/read_bytes/read_text.
        on_write: Fires after write/write_text/write_atomic/open_atomic.
        on_delete: Fires after delete/delete_folder.
        on_copy: Fires after copy.
        on_move: Fires after move.
        on_list: Fires after list_files/list_folders/iter_children/glob/
            get_file_info/get_folder_info/exists/is_file/is_folder/
            resolve.
        on_ping: Fires after ping.
        on_error: Fires on any operation that raises an exception.
        on_any: Fires after every operation (catch-all).
        around: Context-manager factory ``(op, path, backend) -> CM``
            wrapping the entire operation.

    Returns:
        An ``ObservedStore`` proxy.
    """
class BufferedObserver:
    """Collects events and flushes them in batches to a handler.

    Args:
        handler: Called with a list of events on each flush.
        max_queue: Maximum queue size. Events are dropped when full.
        flush_interval: Seconds between automatic flushes.
    """
    def __init__(
        self,
        handler: Callable[[list[StoreEvent]], None],
        *,
        max_queue: int = 1000,
        flush_interval: float = 5.0,
    ) -> None:
    def on_event(self, event: StoreEvent) -> None:
        """Enqueue an event. Drops and warns if queue is full."""
    def flush(self) -> None:
        """Drain the queue and call the handler with collected events."""
    def close(self) -> None:
        """Stop the background thread and perform a final flush."""
```

[61/69] src/remote_store/ext/otel.py (174 rows, definitions)
---
```python
def otel_hooks(
    *,
    tracer_name: str = "remote_store",
    meter_name: str = "remote_store",
    tracer: Tracer | None = None,
    meter: Meter | None = None,
) -> dict[str, Any]:
    """Return hook kwargs for ``observe()``.

    The returned dict contains an ``around`` context-manager hook (tracing)
    and an ``on_any`` callback (metrics).  Unpack it into ``observe()``:

    ```python
    observed = observe(store, **otel_hooks())
    ```

    Args:
        tracer_name: OTel tracer name (default ``"remote_store"``).
            Ignored when *tracer* is provided.
        meter_name: OTel meter name (default ``"remote_store"``).
            Ignored when *meter* is provided.
        tracer: Explicit tracer instance. When ``None`` (default),
            obtained from the global ``TracerProvider`` via *tracer_name*.
        meter: Explicit meter instance. When ``None`` (default),
            obtained from the global ``MeterProvider`` via *meter_name*.

    Returns:
        A dict with ``around`` and ``on_any`` keys.
    """
def otel_observe(
    store: Store,
    *,
    tracer_name: str = "remote_store",
    meter_name: str = "remote_store",
    tracer: Tracer | None = None,
    meter: Meter | None = None,
) -> ObservedStore:
    """Convenience: wrap a Store with OTel tracing + metrics in one call.

    Equivalent to ``observe(store, **otel_hooks(...))``.

    Args:
        store: The Store to observe.
        tracer_name: OTel tracer name (default ``"remote_store"``).
            Ignored when *tracer* is provided.
        meter_name: OTel meter name (default ``"remote_store"``).
            Ignored when *meter* is provided.
        tracer: Explicit tracer instance (see ``otel_hooks()``).
        meter: Explicit meter instance (see ``otel_hooks()``).

    Returns:
        An ``ObservedStore`` with OTel instrumentation.
    """
```

[62/69] src/remote_store/ext/parquet.py (484 rows, definitions)
---
```python
class DatasetIncomplete(RemoteStoreError):
    """Raised when a dataset is structurally incomplete.

    Raised by ``ParquetDatasetStore.read_dataset()`` when the ``_SUCCESS``
    marker is missing or when parts listed in the manifest cannot be found.
    Not ``NotFound`` — some files may exist; the dataset as a whole is not
    in a readable state.
    """
class ManifestCorrupted(RemoteStoreError):
    """Raised when a manifest file cannot be parsed or is structurally invalid.

    Raised by ``ParquetDatasetStore.read_dataset()`` and ``read_manifest()``
    when ``manifest.json`` exists but contains invalid JSON or is missing
    required fields.

    Args:
        reason: The specific parse or structural failure.
    """
    def __init__(
        self,
        message: str = "",
        *,
        path: str | None = None,
        backend: str | None = None,
        reason: str = "",
    ) -> None:
    def __str__(self) -> str:
    def __repr__(self) -> str:
@dataclasses.dataclass(frozen=True)
class DatasetManifest:
    """Immutable metadata record for a written Parquet dataset.

    Args:
        dataset_key: The store-relative prefix under which the dataset lives.
        parts: Relative filenames of the Parquet part files.
        row_count: Total number of rows across all parts.
        schema_hash: First 16 hex characters of the SHA-256 of ``schema.to_string()``.
        compression: Compression codec used (e.g. ``"zstd"``, ``"snappy"``).
        created_at_utc: ISO 8601 timestamp of dataset creation.
        run_id: Optional caller-supplied run identifier for lineage tracking.
        metadata: Optional caller-supplied key-value metadata.
    """
    dataset_key: str
    parts: list[str]
    row_count: int
    schema_hash: str
    compression: str
    created_at_utc: str
    run_id: str | None = None
    metadata: dict[str, str] | None = None
    def to_json(self) -> str:
        """Serialize the manifest to a JSON string.

        Returns:
            A JSON string with sorted keys and 2-space indentation.
        """
    @classmethod
    def from_json(cls, text: str) -> DatasetManifest:
        """Deserialize a manifest from a JSON string.

        Args:
            text: The JSON string to parse.

        Returns:
            A ``DatasetManifest`` instance.

        Raises:
            ManifestCorrupted: If the JSON is invalid, required fields are
                missing, or field types are wrong.
        """
class ParquetDatasetStore:
    """High-level Parquet dataset operations backed by a ``Store``.

    Writes multi-part Parquet datasets with manifest metadata and atomic
    completion markers, enabling reliable dataset exchange over any
    remote-store backend.

    Args:
        store: The Store instance to use for I/O.
        compression: Parquet compression codec. Default ``"zstd"``.
        row_group_size: Maximum rows per row group within each Parquet file.
            ``None`` (default) uses PyArrow's default.
        max_rows_per_file: If set, split the table into multiple part files
            of at most this many rows each. ``None`` writes a single file.
    """
    def __init__(
        self,
        store: Store,
        *,
        compression: str = "zstd",
        row_group_size: int | None = None,
        max_rows_per_file: int | None = None,
    ) -> None:
    @property
    def store(self) -> Store:
        """The underlying store."""
    @property
    def compression(self) -> str:
        """The Parquet compression codec."""
    @property
    def row_group_size(self) -> int | None:
        """Maximum rows per row group, or ``None`` for PyArrow default."""
    @property
    def max_rows_per_file(self) -> int | None:
        """Maximum rows per part file, or ``None`` for single file."""
    def write_dataset(
        self,
        table: pa.Table,
        key: str,
        *,
        overwrite: bool = False,
        run_id: str | None = None,
        metadata: dict[str, str] | None = None,
    ) -> DatasetManifest:
        """Write a PyArrow table as a Parquet dataset under *key*.

        Args:
            table: The PyArrow table to write.
            key: Store-relative prefix for the dataset (e.g. ``"silver/orders"``).
            overwrite: If ``True``, delete any existing dataset at *key* first.
                Requires the ``DELETE`` capability.
            run_id: Optional run identifier for lineage tracking.
            metadata: Optional key-value metadata to embed in the manifest.

        Returns:
            A ``DatasetManifest`` describing the written dataset.

        Raises:
            CapabilityNotSupported: If the store lacks ``ATOMIC_WRITE``, or
                ``DELETE`` when *overwrite* is ``True``.
            AlreadyExists: If *overwrite* is ``False`` and the dataset exists.
        """
    def read_dataset(
        self,
        key: str,
        *,
        columns: list[str] | None = None,
    ) -> pa.Table:
        """Read a Parquet dataset from *key*.

        Args:
            key: Store-relative prefix for the dataset.
            columns: Optional subset of columns to read. ``None`` reads all.

        Returns:
            A PyArrow table with the concatenated contents of all parts.

        Raises:
            DatasetIncomplete: If the ``_SUCCESS`` marker is missing or any
                part listed in the manifest cannot be found.
            ManifestCorrupted: If the manifest cannot be parsed.
            NotFound: If the manifest file does not exist.
        """
    def read_manifest(self, key: str) -> DatasetManifest:
        """Read and parse the manifest for the dataset at *key*.

        Args:
            key: Store-relative prefix for the dataset.

        Returns:
            The parsed ``DatasetManifest``.

        Raises:
            NotFound: If ``manifest.json`` does not exist.
            ManifestCorrupted: If the manifest cannot be parsed.
        """
    def dataset_exists(self, key: str) -> bool:
        """Check whether a completed dataset exists at *key*.

        Args:
            key: Store-relative prefix for the dataset.

        Returns:
            ``True`` if the ``_SUCCESS`` marker exists, ``False`` otherwise.
        """
    def delete_dataset(self, key: str) -> None:
        """Delete an entire dataset at *key*.

        Args:
            key: Store-relative prefix for the dataset.

        Raises:
            NotFound: If the dataset folder does not exist.
        """
```

[63/69] src/remote_store/ext/partition.py (129 rows, definitions)
---
```python
@dataclasses.dataclass(frozen=True)
class ParsedPartition:
    """Result of parsing a Hive-style partition path.

    Attributes:
        partitions: Ordered mapping of partition column names to values.
        filename: The trailing non-partition portion of the path.
    """
    partitions: dict[str, str]
    filename: str
def partition_path(filename: str, /, **partitions: str | int) -> str:
    """Build a Hive-style partition path.

    Args:
        filename: Leaf file name (e.g., ``"data.parquet"``).
            Must be non-empty and must not contain ``/``.
        partitions: Partition key-value pairs. Values are coerced to
            ``str``. Keys and coerced values must be non-empty and
            must not contain ``=``.

    Returns:
        Forward-slash-joined path like ``"year=2026/month=03/data.parquet"``.

    Raises:
        ValueError: If *filename* is empty or contains ``/``, if any
            partition key or coerced value is empty, or if any key or
            value contains ``=``.
    """
def parse_partition(path: str) -> ParsedPartition:
    """Parse a Hive-style partition path into its components.

    A segment is treated as a partition if it contains exactly one ``=``
    and the key portion is non-empty.  Once a non-partition segment is
    encountered, all remaining segments (including any later ``key=value``
    segments) become part of the filename.

    Args:
        path: The partition path to parse (e.g.,
            ``"year=2026/month=03/data.parquet"``).

    Returns:
        A ``ParsedPartition`` with extracted partitions and filename.

    Raises:
        ValueError: If *path* is empty.
    """
```

[64/69] src/remote_store/ext/pydantic.py (74 rows, definitions)
---
```python
def from_pydantic(model: BaseModel) -> RegistryConfig:
    """Convert a Pydantic model to a ``RegistryConfig``.

    Calls ``model.model_dump()`` to produce a plain dict, then delegates to
    ``RegistryConfig.from_dict()``. Secret wrapping, unknown-key warnings,
    and validation all happen in ``from_dict()``.

    Pydantic ``SecretStr`` fields in backend ``options`` dicts are
    automatically unwrapped to plain strings before reaching ``from_dict()``,
    so ``from_dict()``'s sensitive-key detection works correctly. Users may
    use either ``str`` or ``SecretStr`` for credential values in their models.

    Args:
        model: A Pydantic model whose ``model_dump()`` output has
            ``backends`` and ``stores`` keys matching the RegistryConfig schema.

    Returns:
        An immutable ``RegistryConfig``.

    Raises:
        TypeError: If the model dump does not conform to the expected schema.
    """
```

[65/69] src/remote_store/ext/streams.py (215 rows, definitions)
---
```python
class ProgressReader(_StreamWrapper):
    """Readable ``BinaryIO`` wrapper that fires a callback per ``read()``.

    The callback receives the number of bytes read (not cumulative).
    Empty reads do not fire the callback.

    Args:
        inner: Readable binary stream to wrap.
        callback: Called with ``len(data)`` after each non-empty ``read()``.
    """
    def __init__(self, inner: BinaryIO, callback: Callable[[int], None]) -> None:
    def read(self, size: int = -1) -> bytes:
class ProgressWriter(_StreamWrapper):
    """Writable ``BinaryIO`` wrapper that fires a callback per ``write()``.

    The callback receives the number of bytes written (not cumulative).
    Empty writes do not fire the callback.

    Note:
        Assumes buffered I/O semantics — ``write()`` consumes all data or
        raises.  Wrapping a ``RawIOBase`` (partial-write) stream would cause
        the reported byte count to diverge from the bytes actually written.

    Args:
        inner: Writable binary stream to wrap.
        callback: Called with ``len(data)`` after each non-empty ``write()``.
    """
    def __init__(self, inner: BinaryIO, callback: Callable[[int], None]) -> None:
    def write(self, data: bytes | bytearray) -> int:
class ChecksumReader(_StreamWrapper):
    """Readable ``BinaryIO`` wrapper that computes a rolling hash.

    Args:
        inner: Readable binary stream to wrap.
        algorithm: Hash algorithm name (default ``"sha256"``).
            Must be supported by ``hashlib``.
    """
    def __init__(self, inner: BinaryIO, algorithm: str = "sha256") -> None:
    @property
    def algorithm(self) -> str:
        """Hash algorithm name (lowercase)."""
    def read(self, size: int = -1) -> bytes:
    def readline(self, size: int = -1) -> bytes:
    def readlines(self, hint: int = -1) -> list[bytes]:
    def hexdigest(self) -> str:
        """Return the lowercase hex digest of all bytes read so far."""
class ChecksumWriter(_StreamWrapper):
    """Writable ``BinaryIO`` wrapper that computes a rolling hash.

    Note:
        Assumes buffered I/O semantics — ``write()`` consumes all data or
        raises.  Wrapping a ``RawIOBase`` (partial-write) stream would cause
        the hash to include bytes that were not actually written.

    Args:
        inner: Writable binary stream to wrap.
        algorithm: Hash algorithm name (default ``"sha256"``).
            Must be supported by ``hashlib``.
    """
    def __init__(self, inner: BinaryIO, algorithm: str = "sha256") -> None:
    @property
    def algorithm(self) -> str:
        """Hash algorithm name (lowercase)."""
    def write(self, data: bytes | bytearray) -> int:
    def hexdigest(self) -> str:
        """Return the lowercase hex digest of all bytes written so far."""
def read_with_progress(
    store: Store,
    path: str,
    callback: Callable[[int], None],
) -> ProgressReader:
    """Read a file with progress tracking.

    Convenience wrapper around ``ProgressReader(store.read(path), callback)``.
    The caller is responsible for closing the returned stream.

    Args:
        store: The Store to read from.
        path: Store-relative file path.
        callback: Called with byte count after each non-empty ``read()``.

    Returns:
        A ``ProgressReader`` wrapping the store's read stream.
    """
```

[66/69] src/remote_store/ext/transfer.py (151 rows, definitions)
---
```python
def upload(
    store: Store,
    local_path: str | os.PathLike[str],
    remote_path: str,
    *,
    overwrite: bool = False,
    on_progress: Callable[[int], None] | None = None,
) -> None:
    """Upload a local file to a Store.

    Opens the local file in binary read mode and streams it to the Store
    via ``store.write()``. The full file is never loaded into memory.

    Args:
        store: The Store to write to.
        local_path: Path to the local file.
        remote_path: Destination key in the Store.
        overwrite: Forwarded to ``store.write()``.
        on_progress: Called per read with the byte count (not cumulative).

    Returns:
        None

    Raises:
        FileNotFoundError: If *local_path* does not exist.
    """
def download(
    store: Store,
    remote_path: str,
    local_path: str | os.PathLike[str],
    *,
    overwrite: bool = False,
    on_progress: Callable[[int], None] | None = None,
) -> None:
    """Download a file from a Store to a local path.

    Reads the remote file in 1 MiB chunks and writes each chunk to the
    local file. The full file is never loaded into memory.

    Args:
        store: The Store to read from.
        remote_path: Key to read from the Store.
        local_path: Destination path on the local filesystem.
        overwrite: If ``False`` (default) and *local_path* exists,
            raises ``FileExistsError``.
        on_progress: Called per read with the byte count (not cumulative).

    Returns:
        None

    Raises:
        FileExistsError: If *local_path* exists and *overwrite* is False.
    """
def transfer(
    src_store: Store,
    src_path: str,
    dst_store: Store,
    dst_path: str,
    *,
    overwrite: bool = False,
    on_progress: Callable[[int], None] | None = None,
) -> None:
    """Transfer a file from one Store to another.

    Reads the source file and streams it to the destination via
    ``dst_store.write()``. The full file is never loaded into memory.

    Args:
        src_store: The Store to read from.
        src_path: Key to read from *src_store*.
        dst_store: The Store to write to (may be the same as *src_store*).
        dst_path: Destination key in *dst_store*.
        overwrite: Forwarded to ``dst_store.write()``.
        on_progress: Called per read with the byte count (not cumulative).

    Returns:
        None
    """
```

[67/69] src/remote_store/ext/write.py (162 rows, definitions)
---
```python
class HashingAtomicWriter(ChecksumWriter):
    """Writable stream wrapper used by ``open_atomic_with_hash``.

    Subclasses ``ChecksumWriter`` to add a ``.result`` attribute that
    is populated with a ``WriteResult`` after the context manager exits
    successfully.  ``result`` is ``None`` if the block raised.
    """
    result: WriteResult | None
    def __init__(self, inner: BinaryIO, algorithm: str = "sha256") -> None:
    def write(self, data: bytes | bytearray) -> int:
def write_with_hash(
    store: Store,
    path: str,
    content: BinaryIO | bytes,
    *,
    algorithm: str = "sha256",
    overwrite: bool = False,
    metadata: Mapping[str, str] | None = None,
) -> WriteResult:
    """Write *content* to *path* and return a ``WriteResult`` with a client-computed digest.

    Works on every backend declaring ``Capability.WRITE``.
    The hash is always computed client-side regardless of
    ``WRITE_RESULT_NATIVE``.

    Args:
        store: The Store to write to.
        path: Store-relative file path.
        content: ``bytes`` or readable binary stream.
        algorithm: Hash algorithm name (default ``"sha256"``).
        overwrite: If ``False``, raises ``AlreadyExists`` when *path* exists.
        metadata: Optional user metadata (see ``Store.write()``).

    Returns:
        ``WriteResult`` with ``digest`` populated from the client-side hash.
    """
@contextlib.contextmanager
def open_atomic_with_hash(
    store: Store,
    path: str,
    *,
    algorithm: str = "sha256",
    overwrite: bool = False,
    metadata: Mapping[str, str] | None = None,
) -> Iterator[HashingAtomicWriter]:
    """Context manager for streaming atomic writes with client-side hashing.

    Requires ``Capability.ATOMIC_WRITE``.  On successful exit, the
    yielded ``HashingAtomicWriter.result`` is a ``WriteResult`` with
    ``digest`` populated.  On exception ``result`` remains ``None``.

    **Metadata branch:** When *metadata* is non-empty, the implementation
    buffers all written bytes in memory (``io.BytesIO``) and then calls
    ``store.write_atomic()`` on exit.  This means (a) the full payload is
    held in RAM, and (b) validation errors (``ValueError``) and capability
    checks (``CapabilityNotSupported``) are raised after the caller has
    finished writing — not before.  For large payloads, prefer calling
    ``store.write()`` directly with ``metadata=``.

    **``source`` field:** Both branches always set ``source="basic"`` on
    the returned ``WriteResult``.  The metadata branch discards the
    ``source`` returned by ``store.write_atomic()`` because the rich-
    source contract cannot be honored for the no-metadata branch
    (``open_atomic`` returns only a stream, not a ``WriteResult``), so
    both branches use the same value for consistency.

    Args:
        store: The Store to write to.
        path: Store-relative file path.
        algorithm: Hash algorithm name (default ``"sha256"``).
        overwrite: If ``False``, raises ``AlreadyExists`` when *path* exists.
        metadata: Optional user metadata forwarded to ``Store.write_atomic()``
            on backends that declare ``USER_METADATA``.

    Yields:
        ``HashingAtomicWriter`` — write to it as a binary stream.

    Raises:
        CapabilityNotSupported: If the backend lacks ``ATOMIC_WRITE``.
        AlreadyExists: If *path* exists and *overwrite* is ``False``.
    """
```

[68/69] src/remote_store/ext/yaml.py (99 rows, definitions)
---
```python
def from_yaml(
    path: str | Path,
    *,
    resolve_env_vars: bool = False,
) -> RegistryConfig:
    """Load config from a YAML file.

    Accepts either ``pyyaml`` or ``ruamel.yaml`` as the parser.

    Args:
        path: Path to the YAML file.
        resolve_env_vars: When ``True``, resolve ``${VAR}`` placeholders
            via ``resolve_env`` before constructing
            the config.

    Returns:
        An immutable ``RegistryConfig``.

    Raises:
        ModuleNotFoundError: If neither ``pyyaml`` nor ``ruamel.yaml``
            is installed.
        FileNotFoundError: If *path* does not exist.
        TypeError: If the top-level YAML value is not a mapping.
        KeyError: If *resolve_env_vars* is ``True`` and a placeholder
            references an unset variable with no default.
    """
```

[69/69] src/remote_store/py.typed (0 rows)