From dc954dad4bc32bbb92a231fa610171b6ee7a7560 Mon Sep 17 00:00:00 2001
From: terminalchai <213856599+terminalchai@users.noreply.github.com>
Date: Sat, 10 Oct 2026 01:39:11 +0530
Subject: [PATCH 1/2] fix: clear stale caches after cloud mutations

---
 HISTORY.md                |   2 +
 cloudpathlib/cloudpath.py |   5 ++
 tests/test_caching.py     | 104 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 111 insertions(+)

diff --git a/HISTORY.md b/HISTORY.md
index 3bdd04eb..fa3e3fc8 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -5,6 +5,8 @@
 - Added streaming I/O via `FileCacheMode.streaming` for S3, Azure Blob Storage, Google Cloud Storage, and HTTP/HTTPS. In this mode `CloudPath.open()` returns the standard `io` file objects (`BufferedReader`/`BufferedWriter`/`TextIOWrapper`) backed by ranged reads and multipart uploads instead of the local cache, so only the bytes you read are downloaded and written data is uploaded as you write it. Read streams are seekable; append and update modes fall back to the cache and remove the cache file on close. `buffering` has its builtin `open()` meaning and sets the ranged-request size (default 5 MiB). A new `streaming_max_concurrency` client option (default 4) bounds background read-ahead and part uploads per stream; the buffer size, concurrency, and multipart part size can also be set with the `CLOUDPATHLIB_STREAMING_BUFFER_SIZE`, `CLOUDPATHLIB_STREAMING_MAX_CONCURRENCY`, and `CLOUDPATHLIB_[<PROVIDER>_]STREAMING_PART_SIZE` environment variables. Opening a file for append or update in streaming mode warns that it falls back to the cache. `copy`/`rename`/`replace` stream between clients, `fspath` raises `CloudPathNotImplementedError`, and failures raise `cloudpathlib.exceptions` types including the new `CloudPathStreamingError`. See the [Streaming I/O](https://cloudpathlib.drivendata.org/stable/streaming_io/) docs. (PR [#535](https://github.com/drivendataorg/cloudpathlib/pull/535))
 - Changed append mode (`a`, `ab`, `a+`, and `ab+`) to create missing cloud files, matching Python file behavior. Appending to existing files still preserves their content. (Issue [#585](https://github.com/drivendataorg/cloudpathlib/issues/585))
 
+- Fixed stale local caches after `unlink`, `rename`, `replace`, `rmdir`, and `rmtree`. Deleted files can be recreated with exclusive or append mode, and moved files no longer reuse cached destination content. (Issue [#388](https://github.com/drivendataorg/cloudpathlib/issues/388))
+
 ## v0.26.0 (2026-10-01)
 
 - Added support for Python 3.15: `CloudPath.mkdir(parent_mode=...)` works across providers, and cloud paths can be pickled repeatedly without a `ParseResult` error. (Issue [#593](https://github.com/drivendataorg/cloudpathlib/issues/593), PR [#594](https://github.com/drivendataorg/cloudpathlib/pull/594))
diff --git a/cloudpathlib/cloudpath.py b/cloudpathlib/cloudpath.py
index b1d552ef..0df463c3 100644
--- a/cloudpathlib/cloudpath.py
+++ b/cloudpathlib/cloudpath.py
@@ -1010,6 +1010,8 @@ def replace(self, target: Self) -> Self:
             target.unlink()
 
         self.client._move_file(self, target)
+        self.clear_cache()
+        target.clear_cache()
         return target
 
     def rename(self, target: Self) -> Self:
@@ -1030,6 +1032,7 @@ def rmdir(self) -> None:
         except StopIteration:
             pass
         self.client._remove(self)
+        self.clear_cache()
 
     def samefile(self, other_path: Union[str, os.PathLike]) -> bool:
         # all cloud paths are absolute and the paths are used for hash
@@ -1042,6 +1045,7 @@ def unlink(self, missing_ok: bool = True) -> None:
                 f"Path {self} is a directory; call rmdir instead of unlink."
             )
         self.client._remove(self, missing_ok)
+        self.clear_cache()
 
     def write_bytes(self, data: bytes) -> int:
         """Open the file in bytes mode, write to it, and close the file.
@@ -1322,6 +1326,7 @@ def rmtree(self) -> None:
                 f"Path {self} is a file; call unlink instead of rmtree."
             )
         self.client._remove(self)
+        self.clear_cache()
 
     def upload_from(
         self,
diff --git a/tests/test_caching.py b/tests/test_caching.py
index 2bc93658..b500e7df 100644
--- a/tests/test_caching.py
+++ b/tests/test_caching.py
@@ -580,3 +580,107 @@ def test_streaming_append_fallback_cache_cleaned_up(rig: CloudProviderTestRig):
 
     assert not cp._local.exists()
     assert cp.read_text() == original + "appended"
+
+
+@pytest.mark.parametrize("mode", ["x", "a"])
+def test_unlink_clears_cache_before_recreating(rig: CloudProviderTestRig, mode):
+    cp = rig.create_cloud_path("unlink-cache.txt")
+    cp.write_text("old content")
+    assert cp._local.exists()
+
+    cp.unlink()
+
+    assert not cp.exists()
+    assert not cp._local.exists()
+    with cp.open(mode) as handle:
+        handle.write("new content")
+    assert cp.read_text() == "new content"
+
+
+@pytest.mark.parametrize("method", ["rename", "replace"])
+@pytest.mark.parametrize("target_exists", [False, True])
+def test_move_clears_source_and_target_caches(rig: CloudProviderTestRig, method, target_exists):
+    source = rig.create_cloud_path("move-source.txt")
+    target = rig.create_cloud_path("move-target.txt")
+    source.write_text("source content")
+    target.write_text("old target content")
+    if not target_exists:
+        # Model a remote deletion made outside cloudpathlib, leaving a stale target cache.
+        target.client._remove(target)
+    assert source._local.exists()
+    assert target._local.exists()
+
+    result = getattr(source, method)(target)
+
+    assert result == target
+    assert not source.exists()
+    assert not source._local.exists()
+    assert not target._local.exists()
+    assert target.read_text() == "source content"
+    with source.open("x") as handle:
+        handle.write("recreated source")
+    assert source.read_text() == "recreated source"
+
+
+@pytest.mark.parametrize("method", ["rename", "replace"])
+def test_move_to_same_path_preserves_cache(rig: CloudProviderTestRig, method):
+    cp = rig.create_cloud_path("same-path.txt")
+    cp.write_text("unchanged")
+
+    assert getattr(cp, method)(cp) == cp
+
+    assert cp._local.exists()
+    assert cp.read_text() == "unchanged"
+
+
+def test_rmtree_clears_only_its_cache_subtree(rig: CloudProviderTestRig):
+    directory = rig.create_cloud_path("delete-tree/")
+    child = directory / "nested" / "file.txt"
+    sibling = rig.create_cloud_path("keep-tree/file.txt")
+    child.write_text("delete")
+    sibling.write_text("keep")
+    assert child._local.exists()
+    assert sibling._local.exists()
+
+    directory.rmtree()
+
+    assert not directory.exists()
+    assert not directory._local.exists()
+    assert sibling._local.exists()
+    assert sibling.read_text() == "keep"
+
+
+def test_rmdir_clears_stale_cached_children(rig: CloudProviderTestRig, monkeypatch):
+    directory = rig.create_cloud_path("empty-directory/")
+    child = directory / "file.txt"
+    child.write_text("stale")
+    child.client._remove(child)
+    assert child._local.exists()
+    # Object stores need not retain an empty directory; some mocks cannot list it.
+    monkeypatch.setattr(directory, "iterdir", lambda: iter(()))
+
+    directory.rmdir()
+
+    assert not directory._local.exists()
+
+
+@pytest.mark.parametrize("method", ["unlink", "rename"])
+def test_failed_mutation_preserves_source_cache(rig: CloudProviderTestRig, method, monkeypatch):
+    cp = rig.create_cloud_path("failed-mutation.txt")
+    cp.write_text("preserved")
+
+    def fail(*args, **kwargs):
+        raise RuntimeError("simulated cloud failure")
+
+    if method == "unlink":
+        monkeypatch.setattr(cp.client, "_remove", fail)
+        args = ()
+    else:
+        monkeypatch.setattr(cp.client, "_move_file", fail)
+        args = (rig.create_cloud_path("failed-move-target.txt"),)
+
+    with pytest.raises(RuntimeError, match="simulated cloud failure"):
+        getattr(cp, method)(*args)
+
+    assert cp.exists()
+    assert cp._local.read_text() == "preserved"

From b08c4273e12e225fd56d367865c84e7ddf6a1425 Mon Sep 17 00:00:00 2001
From: terminalchai <213856599+terminalchai@users.noreply.github.com>
Date: Tue, 13 Oct 2026 02:59:14 +0530
Subject: [PATCH 2/2] fix: preserve pending writes across cache invalidation
 paths

---
 HISTORY.md                       |   2 +-
 cloudpathlib/azure/azblobpath.py |   8 +-
 cloudpathlib/client.py           |   3 +
 cloudpathlib/cloudpath.py        |  82 ++++++-
 tests/test_caching.py            | 370 +++++++++++++++++++++++++++++++
 5 files changed, 457 insertions(+), 8 deletions(-)

diff --git a/HISTORY.md b/HISTORY.md
index fa3e3fc8..8eef1ff1 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -5,7 +5,7 @@
 - Added streaming I/O via `FileCacheMode.streaming` for S3, Azure Blob Storage, Google Cloud Storage, and HTTP/HTTPS. In this mode `CloudPath.open()` returns the standard `io` file objects (`BufferedReader`/`BufferedWriter`/`TextIOWrapper`) backed by ranged reads and multipart uploads instead of the local cache, so only the bytes you read are downloaded and written data is uploaded as you write it. Read streams are seekable; append and update modes fall back to the cache and remove the cache file on close. `buffering` has its builtin `open()` meaning and sets the ranged-request size (default 5 MiB). A new `streaming_max_concurrency` client option (default 4) bounds background read-ahead and part uploads per stream; the buffer size, concurrency, and multipart part size can also be set with the `CLOUDPATHLIB_STREAMING_BUFFER_SIZE`, `CLOUDPATHLIB_STREAMING_MAX_CONCURRENCY`, and `CLOUDPATHLIB_[<PROVIDER>_]STREAMING_PART_SIZE` environment variables. Opening a file for append or update in streaming mode warns that it falls back to the cache. `copy`/`rename`/`replace` stream between clients, `fspath` raises `CloudPathNotImplementedError`, and failures raise `cloudpathlib.exceptions` types including the new `CloudPathStreamingError`. See the [Streaming I/O](https://cloudpathlib.drivendata.org/stable/streaming_io/) docs. (PR [#535](https://github.com/drivendataorg/cloudpathlib/pull/535))
 - Changed append mode (`a`, `ab`, `a+`, and `ab+`) to create missing cloud files, matching Python file behavior. Appending to existing files still preserves their content. (Issue [#585](https://github.com/drivendataorg/cloudpathlib/issues/585))
 
-- Fixed stale local caches after `unlink`, `rename`, `replace`, `rmdir`, and `rmtree`. Deleted files can be recreated with exclusive or append mode, and moved files no longer reuse cached destination content. (Issue [#388](https://github.com/drivendataorg/cloudpathlib/issues/388))
+- Fixed stale local caches after cloud file and directory deletions and moves, including `move`, `move_into`, and ADLS Gen2 directory renames. Clean cached files are also discarded after external cloud deletions, so exclusive creation and append start fresh; pending local changes are preserved when the cloud object is already absent, including through equivalent paths on the same client. Moves with pending cached writes raise `OverwriteDirtyFileError` before modifying cloud objects. (Issue [#388](https://github.com/drivendataorg/cloudpathlib/issues/388), PR [#600](https://github.com/drivendataorg/cloudpathlib/pull/600))
 
 ## v0.26.0 (2026-10-01)
 
diff --git a/cloudpathlib/azure/azblobpath.py b/cloudpathlib/azure/azblobpath.py
index 1c5c4bb0..8d8601e4 100644
--- a/cloudpathlib/azure/azblobpath.py
+++ b/cloudpathlib/azure/azblobpath.py
@@ -87,7 +87,13 @@ def replace(self, target: "AzureBlobPath") -> "AzureBlobPath":
         # we can rename directories on ADLS Gen2
         except CloudPathIsADirectoryError:
             if self.client._check_hns(self):
-                return self.client._move_file(self, target)
+                if target == self:
+                    return self
+                self._raise_if_pending_move_cache(target)
+                result = self.client._move_file(self, target)
+                self.clear_cache()
+                target.clear_cache()
+                return result
             else:
                 raise
 
diff --git a/cloudpathlib/client.py b/cloudpathlib/client.py
index d5c25948..e2e3cf1a 100644
--- a/cloudpathlib/client.py
+++ b/cloudpathlib/client.py
@@ -71,6 +71,8 @@ def __init__(
     ) -> None:
         self.file_cache_mode = None
         self._cache_tmp_dir = None
+        # Writer identities avoid retaining CloudPath objects and distinguish aliases.
+        self._dirty_cache_files: Dict[Path, set[int]] = {}
         self._cloud_meta.validate_completeness()
 
         # concurrent requests per open streaming file (part uploads / read-ahead);
@@ -174,6 +176,7 @@ def clear_cache(self):
                     p.unlink()
                 else:
                     shutil.rmtree(p)
+        self._dirty_cache_files.clear()
 
     @abc.abstractmethod
     def _download_file(
diff --git a/cloudpathlib/cloudpath.py b/cloudpathlib/cloudpath.py
index 0df463c3..87d320d6 100644
--- a/cloudpathlib/cloudpath.py
+++ b/cloudpathlib/cloudpath.py
@@ -900,6 +900,10 @@ def _patched_close_upload(*args, **kwargs) -> None:
                 # we already ran our close method patch
                 if not self._dirty:
                     return
+                if id(self) not in self.client._dirty_cache_files.get(self._local, set()):
+                    # Explicit cleanup discarded these pending writes; do not retry them.
+                    self._dirty = False
+                    return
 
                 # original mtime should match what was in the cloud; because of system clocks or rounding
                 # by the cloud provider, the new version in our cache is not newer than the original
@@ -918,7 +922,8 @@ def _patched_close_upload(*args, **kwargs) -> None:
             # keep reference in case we need to close when __del__ is called on this object
             self._handle = buffer
 
-            # opened for write, so mark dirty
+            # Track pending writes by cache file, since equivalent paths share it.
+            self.client._dirty_cache_files.setdefault(self._local, set()).add(id(self))
             self._dirty = True
 
         # if we don't want any cache around, remove the cache as soon as the file is
@@ -992,6 +997,11 @@ def _raise_if_copy_target_newer(
             )
 
     def replace(self, target: Self) -> Self:
+        """Move the cloud object to target, clearing obsolete caches after success.
+
+        Pending cached source or target writes must be uploaded or explicitly
+        cleared first; otherwise :class:`OverwriteDirtyFileError` is raised.
+        """
         if type(self) is not type(target):
             raise TypeError(
                 f"The target based to rename must be an instantiated class of type: {type(self)}"
@@ -1006,6 +1016,7 @@ def replace(self, target: Self) -> Self:
             # Request is to replace/rename this with the same path - nothing to do
             return self
 
+        self._raise_if_pending_move_cache(target)
         if target.exists():
             target.unlink()
 
@@ -1044,8 +1055,12 @@ def unlink(self, missing_ok: bool = True) -> None:
             raise CloudPathIsADirectoryError(
                 f"Path {self} is a directory; call rmdir instead of unlink."
             )
+        # A missing cloud object may still have pending local changes to upload.
+        dirty_cache = self._cache_is_dirty()
+        existed = dirty_cache and self.exists()
         self.client._remove(self, missing_ok)
-        self.clear_cache()
+        if existed or not dirty_cache:
+            self.clear_cache()
 
     def write_bytes(self, data: bytes) -> int:
         """Open the file in bytes mode, write to it, and close the file.
@@ -1380,9 +1395,17 @@ def _copy(
             if destination.exists() and destination.is_dir():
                 destination = destination / self.name
 
+            if remove_src:
+                if destination == self:
+                    return self
+                self._raise_if_pending_move_cache(destination)
             self._raise_if_copy_target_newer(destination, force_overwrite_to_cloud)
 
-            return cast(Self, self.client._move_file(self, destination, remove_src=remove_src))
+            result = self.client._move_file(self, destination, remove_src=remove_src)
+            if remove_src:
+                self.clear_cache()
+                destination.clear_cache()
+            return cast(Self, result)
 
         else:
             if not destination.exists() or destination.is_file():
@@ -1609,7 +1632,11 @@ def move(
         preserve_metadata: bool = False,
         force_overwrite_to_cloud: Optional[bool] = None,
     ) -> Union[Path, Self]:
-        """Move self to target location, removing the source."""
+        """Move self to target location, removing the source.
+
+        Cloud-native moves raise :class:`OverwriteDirtyFileError` before changing
+        cloud objects if source or target caches have pending writes.
+        """
         return self._copy(
             target,
             follow_symlinks=follow_symlinks,
@@ -1664,6 +1691,33 @@ def move_into(
         )
         return cast(Union[Path, Self], result)
 
+    def _raise_if_pending_move_cache(self, target: "CloudPath") -> None:
+        for path in (self, target):
+            pending = path._cache_is_dirty()
+            if not pending:
+                try:
+                    local = path._local
+                    pending = any(
+                        cached_file.is_relative_to(local) and cached_file.is_file()
+                        for cached_file in tuple(path.client._dirty_cache_files)
+                    )
+                except CloudPathLocalPathTraversalError:
+                    pass
+            if pending:
+                raise OverwriteDirtyFileError(
+                    f"Cache for {path} has pending writes. Upload the local changes "
+                    "or explicitly clear the cache before moving."
+                )
+
+    def _cache_is_dirty(self) -> bool:
+        try:
+            local = self._local
+            # Explicit cleanup releases tracking even if an old writer's flag remains set.
+            return local.is_file() and bool(self.client._dirty_cache_files.get(local))
+        except CloudPathLocalPathTraversalError:
+            # A path that cannot map to the cache cannot have pending cache writes.
+            return False
+
     def clear_cache(self):
         """Removes cache if it exists"""
         try:
@@ -1672,11 +1726,18 @@ def clear_cache(self):
             # a key that cannot be safely mapped to a local path can never have been cached,
             # so there is nothing to clear (and __del__ must not raise on such paths)
             return
+        is_directory = local.is_dir()
         if local.exists():
             if local.is_file():
                 local.unlink()
             else:
                 shutil.rmtree(local)
+        if is_directory:
+            for cached_file in tuple(self.client._dirty_cache_files):
+                if cached_file.is_relative_to(local):
+                    self.client._dirty_cache_files.pop(cached_file, None)
+        else:
+            self.client._dirty_cache_files.pop(local, None)
 
     # ===========  private cloud methods ===============
     @property
@@ -1711,8 +1772,10 @@ def _refresh_cache(self, force_overwrite_from_cloud: Optional[bool] = None) -> N
         try:
             stats = self.stat()
         except NoStatError:
-            # nothing to cache if the file does not exist; happens when creating
-            # new files that will be uploaded
+            # Discard stale files after external deletions, retaining pending writes
+            # and cached directory contents that cannot be described by a file stat.
+            if not self._cache_is_dirty() and self._local.is_file():
+                self.clear_cache()
             return
 
         # if not exist or cloud newer
@@ -1764,6 +1827,13 @@ def _upload_local_to_cloud(
         stats = self.stat()
         os.utime(self._local, times=(stats.st_mtime, stats.st_mtime))
 
+        # Remove this writer only; another path instance may still have pending work.
+        writers = self.client._dirty_cache_files.get(self._local)
+        if writers is not None:
+            writers.discard(id(self))
+            if not writers:
+                self.client._dirty_cache_files.pop(self._local, None)
+
         # reset dirty and handle now that this is uploaded
         self._dirty = False
         self._handle = None
diff --git a/tests/test_caching.py b/tests/test_caching.py
index b500e7df..a6bb89ec 100644
--- a/tests/test_caching.py
+++ b/tests/test_caching.py
@@ -13,7 +13,9 @@
 from cloudpathlib.enums import FileCacheMode
 from cloudpathlib.exceptions import (
     InvalidConfigurationException,
+    NoStatError,
     OverwriteNewerCloudError,
+    OverwriteDirtyFileError,
     OverwriteNewerLocalError,
 )
 from tests.conftest import CloudProviderTestRig
@@ -684,3 +686,371 @@ def fail(*args, **kwargs):
 
     assert cp.exists()
     assert cp._local.read_text() == "preserved"
+
+
+@pytest.mark.parametrize("method", ["move", "move_into"])
+def test_native_move_invalidates_both_caches(rig: CloudProviderTestRig, method):
+    source = rig.create_cloud_path("native-move-source.txt")
+    target = rig.create_cloud_path(
+        "native-move-target/" if method == "move_into" else "native-move-target.txt"
+    )
+    destination = target / source.name if method == "move_into" else target
+    source.write_text("source content")
+    destination.write_text("old target content")
+    # A stale destination cache must be invalidated even if its remote file is gone.
+    destination.client._remove(destination)
+    assert source.client is destination.client
+    assert source._local.exists()
+    assert destination._local.exists()
+
+    result = getattr(source, method)(target, force_overwrite_to_cloud=True)
+
+    assert result == destination
+    assert not source.exists()
+    assert not source._local.exists()
+    assert not destination._local.exists()
+    assert result.read_text() == "source content"
+    with source.open("x") as handle:
+        handle.write("recreated")
+    assert source.read_text() == "recreated"
+
+
+@pytest.mark.parametrize("mode", ["x", "a"])
+def test_external_deletion_discards_clean_cache(rig: CloudProviderTestRig, mode):
+    cp = rig.create_cloud_path("externally-deleted.txt")
+    cp.write_text("old content")
+    cp.client._remove(cp)
+    assert cp._local.exists()
+
+    with cp.open(mode) as handle:
+        handle.write("new content")
+
+    assert cp.read_text() == "new content"
+
+
+@pytest.mark.parametrize("existed", [False, True])
+def test_unlink_preserves_only_missing_dirty_cache(
+    rig: CloudProviderTestRig, existed, monkeypatch
+):
+    cp = rig.create_cloud_path("dirty-unlink.txt")
+    if existed:
+        cp.write_text("cloud content")
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(cp.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            cp.write_text("pending local content")
+    assert cp._dirty
+    # The failed upload closed the buffer; avoid retrying it during fixture cleanup.
+    assert cp._handle.closed
+    cp._handle = None
+
+    cp.unlink()
+
+    assert not cp.exists()
+    if existed:
+        assert not cp._local.exists()
+    else:
+        assert cp._local.read_text() == "pending local content"
+
+
+def test_refresh_preserves_missing_dirty_cache(rig: CloudProviderTestRig, monkeypatch):
+    cp = rig.create_cloud_path("dirty-refresh.txt")
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(cp.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            cp.write_text("pending local content")
+    assert cp._dirty
+    # The failed upload closed the buffer; avoid retrying it during fixture cleanup.
+    assert cp._handle.closed
+    cp._handle = None
+
+    cp._refresh_cache()
+
+    assert cp._local.read_text() == "pending local content"
+
+
+def test_refresh_preserves_directory_cache(rig: CloudProviderTestRig, monkeypatch):
+    directory = rig.create_cloud_path("refresh-directory/")
+    child = directory / "file.txt"
+    child.write_text("keep cached child")
+
+    def no_stat():
+        raise NoStatError("directory has no file stat")
+
+    monkeypatch.setattr(directory, "stat", no_stat)
+    directory._refresh_cache()
+
+    assert child._local.read_text() == "keep cached child"
+
+
+@pytest.mark.parametrize("method", ["rename", "replace"])
+def test_adls_directory_move_invalidates_nested_caches(azure_gen2_rig, method):
+    source = azure_gen2_rig.create_cloud_path("cached-directory")
+    source.mkdir()
+    (source / "nested").mkdir()
+    child = source / "nested" / "file.txt"
+    child.write_text("nested source content")
+    target = azure_gen2_rig.create_cloud_path("renamed-directory")
+    target._local.mkdir(parents=True, exist_ok=True)
+    (target._local / "stale.txt").write_text("stale destination cache")
+    assert child._local.exists()
+
+    result = getattr(source, method)(target)
+
+    assert result == target
+    assert not source.exists()
+    assert not source._local.exists()
+    assert not target._local.exists()
+    assert (target / "nested" / "file.txt").read_text() == "nested source content"
+
+
+@pytest.mark.parametrize("method", ["move", "move_into"])
+def test_failed_native_move_preserves_both_caches(rig: CloudProviderTestRig, method, monkeypatch):
+    source = rig.create_cloud_path("failed-native-source.txt")
+    target = rig.create_cloud_path(
+        "failed-native-target/" if method == "move_into" else "failed-native-target.txt"
+    )
+    destination = target / source.name if method == "move_into" else target
+    source.write_text("source content")
+    destination.write_text("target content")
+
+    def fail(*args, **kwargs):
+        raise RuntimeError("simulated move failure")
+
+    monkeypatch.setattr(source.client, "_move_file", fail)
+    with pytest.raises(RuntimeError, match="simulated move failure"):
+        getattr(source, method)(target, force_overwrite_to_cloud=True)
+
+    assert source.exists()
+    assert destination.exists()
+    assert source._local.read_text() == "source content"
+    assert destination._local.read_text() == "target content"
+
+
+def test_failed_adls_directory_move_preserves_caches(azure_gen2_rig, monkeypatch):
+    source = azure_gen2_rig.create_cloud_path("failed-directory")
+    source.mkdir()
+    child = source / "file.txt"
+    child.write_text("cached source content")
+    target = azure_gen2_rig.create_cloud_path("failed-directory-target")
+    target._local.mkdir(parents=True, exist_ok=True)
+    stale_target = target._local / "stale.txt"
+    stale_target.write_text("cached target content")
+
+    def fail(*args, **kwargs):
+        raise RuntimeError("simulated directory move failure")
+
+    monkeypatch.setattr(source.client, "_move_file", fail)
+    with pytest.raises(RuntimeError, match="simulated directory move failure"):
+        source.rename(target)
+
+    assert source.exists()
+    assert child._local.read_text() == "cached source content"
+    assert stale_target.read_text() == "cached target content"
+
+
+@pytest.mark.parametrize("operation", ["refresh", "unlink"])
+def test_clean_alias_preserves_pending_cache(rig: CloudProviderTestRig, operation, monkeypatch):
+    owner = rig.create_cloud_path("aliased-pending.txt")
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(owner.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            owner.write_text("pending local content")
+    assert owner._dirty
+    assert owner._handle.closed
+    owner._handle = None
+    alias = rig.create_cloud_path("aliased-pending.txt", client=owner.client)
+    assert alias is not owner
+    assert not alias._dirty
+    assert alias._local == owner._local
+
+    if operation == "refresh":
+        alias._refresh_cache()
+    else:
+        alias.unlink()
+
+    assert owner._local.read_text() == "pending local content"
+    # The original owner can still retry the failed upload successfully.
+    owner._upload_local_to_cloud()
+    assert owner.read_text() == "pending local content"
+
+
+@pytest.mark.parametrize("cleanup", ["path", "client"])
+def test_explicit_cache_cleanup_releases_pending_tracking(
+    rig: CloudProviderTestRig, cleanup, monkeypatch
+):
+    owner = rig.create_cloud_path("clear-pending-cache.txt")
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(owner.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            owner.write_text("pending local content")
+    assert owner._handle.closed
+    owner._handle = None
+    alias = rig.create_cloud_path("clear-pending-cache.txt", client=owner.client)
+    assert alias._cache_is_dirty()
+
+    if cleanup == "path":
+        alias.clear_cache()
+    else:
+        alias.client.clear_cache()
+
+    assert not alias._cache_is_dirty()
+    with alias.open("x") as handle:
+        handle.write("fresh content")
+    assert not alias._cache_is_dirty()
+    assert alias.read_text() == "fresh content"
+
+
+def test_upload_keeps_other_writer_protection(rig: CloudProviderTestRig, monkeypatch):
+    owners = [rig.create_cloud_path("multiple-pending.txt") for _ in range(2)]
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    for index, owner in enumerate(owners):
+        with monkeypatch.context() as patch:
+            patch.setattr(owner.client, "_upload_file", fail_upload)
+            with pytest.raises(RuntimeError, match="simulated upload failure"):
+                owner.write_text(f"pending {index}")
+        assert owner._handle.closed
+        owner._handle = None
+    alias = rig.create_cloud_path("multiple-pending.txt", client=owners[0].client)
+
+    owners[0]._upload_local_to_cloud()
+    alias.client._remove(alias)
+    alias._refresh_cache()
+    assert alias._local.read_text() == "pending 1"
+
+    owners[1]._upload_local_to_cloud()
+    alias.client._remove(alias)
+    alias._refresh_cache()
+    assert not alias._local.exists()
+
+
+@pytest.mark.parametrize("method", ["rename", "replace", "move", "move_into"])
+@pytest.mark.parametrize("pending_side", ["source", "destination"])
+def test_move_rejects_pending_cached_writes(
+    rig: CloudProviderTestRig, method, pending_side, monkeypatch
+):
+    source = rig.create_cloud_path("pending-move-source.txt")
+    target = rig.create_cloud_path(
+        "pending-move-target/" if method == "move_into" else "pending-move-target.txt"
+    )
+    destination = target / source.name if method == "move_into" else target
+    source.write_text("cloud source")
+    destination.write_text("cloud destination")
+    owner_key = (
+        "pending-move-source.txt"
+        if pending_side == "source"
+        else (
+            "pending-move-target/pending-move-source.txt"
+            if method == "move_into"
+            else "pending-move-target.txt"
+        )
+    )
+    owner = rig.create_cloud_path(owner_key, client=source.client)
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(owner.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            owner.write_text("pending local changes")
+    assert owner._handle.closed
+    owner._handle = None
+    kwargs = {"force_overwrite_to_cloud": True} if method in ("move", "move_into") else {}
+
+    with pytest.raises(OverwriteDirtyFileError, match="pending writes"):
+        getattr(source, method)(target, **kwargs)
+
+    assert source.exists()
+    assert destination.exists()
+    assert owner._local.read_text() == "pending local changes"
+    owner._upload_local_to_cloud(force_overwrite_to_cloud=True)
+    assert owner.read_text() == "pending local changes"
+
+
+@pytest.mark.parametrize("pending_side", ["source", "destination"])
+def test_adls_directory_move_rejects_pending_descendants(
+    azure_gen2_rig, pending_side, monkeypatch
+):
+    source = azure_gen2_rig.create_cloud_path("pending-directory-source")
+    target = azure_gen2_rig.create_cloud_path("pending-directory-target")
+    source.mkdir()
+    target.mkdir()
+    owner = (source if pending_side == "source" else target) / "file.txt"
+    owner.write_text("cloud content")
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(owner.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            owner.write_text("pending directory content")
+    assert owner._handle.closed
+    owner._handle = None
+
+    with pytest.raises(OverwriteDirtyFileError, match="pending writes"):
+        source.rename(target)
+
+    assert source.exists()
+    assert target.exists()
+    assert owner._local.read_text() == "pending directory content"
+
+
+def test_adls_directory_self_rename_preserves_cache(azure_gen2_rig):
+    directory = azure_gen2_rig.create_cloud_path("same-directory")
+    directory.mkdir()
+    child = directory / "file.txt"
+    child.write_text("cached contents")
+
+    assert directory.rename(directory) == directory
+    assert child._local.read_text() == "cached contents"
+
+
+@pytest.mark.parametrize("cleanup", ["path", "client"])
+def test_original_writer_can_move_after_explicit_cleanup(
+    rig: CloudProviderTestRig, cleanup, monkeypatch
+):
+    source = rig.create_cloud_path("cleanup-then-move.txt")
+    source.write_text("cloud content")
+
+    def fail_upload(*args, **kwargs):
+        raise RuntimeError("simulated upload failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(source.client, "_upload_file", fail_upload)
+        with pytest.raises(RuntimeError, match="simulated upload failure"):
+            source.write_text("discarded pending content")
+    assert source._handle.closed
+
+    if cleanup == "path":
+        source.clear_cache()
+    else:
+        source.client.clear_cache()
+    # A retry of the closed handle must not re-upload explicitly discarded data.
+    source._handle.close()
+    target = rig.create_cloud_path("cleanup-move-target.txt")
+    result = source.rename(target)
+
+    assert result == target
+    assert not source.exists()
+    assert target.read_text() == "cloud content"