fix: bound Windows metadata replacement retries

docs/platform-support.md

@@ -66,3 +66,11 @@ provided by the Windows mount and are outside the Linux filesystem contract.
 WSL2 support does not imply that the generic package translates paths between
 Linux and Windows or that a consumer's native Windows commands are available
 inside the distribution.
+
+On native Windows, private metadata replacement retries sharing-violation and
+lock-violation errors (`winerror` 32 and 33). A Windows access-denied response
+is retried only when the source follows base-cli's own temporary-file naming
+contract, which covers an in-use destination reported as `winerror` 5; other
+access-denied and permanent permission/path errors fail immediately. Transient
+retries are bounded by a one-second elapsed deadline; the destination remains
+untouched if that deadline is exhausted.

lib/python/base_cli/_private_files.py

@@ -12,6 +12,9 @@
 
 PRIVATE_FILE_MODE = 0o600
 PRIVATE_DIRECTORY_MODE = 0o700
+_WINDOWS_REPLACE_RETRY_DEADLINE_SECONDS = 1.0
+_WINDOWS_REPLACE_INITIAL_DELAY_SECONDS = 0.005
+_WINDOWS_RETRYABLE_WINERRORS = frozenset({32, 33})  # sharing and lock violations
 
 
 def restrict_file(path: Path) -> None:
@@ -182,16 +185,35 @@ def _sync_directory(parent_fd: int) -> None:
 def _replace_with_retry(source: Path, destination: Path) -> None:
     """Replace a private file, tolerating transient Windows sharing races."""
 
+    if os.name != "nt":
+        os.replace(source, destination)
+        return
+
     # Antivirus/indexer handles and concurrent writers can hold the destination
-    # briefly on Windows. Use a bounded, linear backoff long enough for those
-    # transient sharing violations without making a persistent permission error
-    # unbounded.
-    attempts = 1 if os.name != "nt" else 50
-    for attempt in range(attempts):
+    # briefly on Windows. Retry only documented sharing/lock violations and
+    # bound the total delay so permanent ACL/path failures remain actionable.
+    deadline = time.monotonic() + _WINDOWS_REPLACE_RETRY_DEADLINE_SECONDS
+    attempt = 0
+    while True:
         try:
             os.replace(source, destination)
             return
-        except PermissionError:
-            if attempt == attempts - 1:
+        except PermissionError as exc:
+            winerror = getattr(exc, "winerror", None)
+            retryable = winerror in _WINDOWS_RETRYABLE_WINERRORS
+            # Windows can report an in-use destination as WinError 5 when the
+            # competing process has opened it without sharing. Restrict this
+            # compatibility case to the temporary-file naming contract owned
+            # by this helper; arbitrary access-denied operations still fail
+            # immediately.
+            retryable = retryable or (
+                winerror == 5 and source.name.startswith(f".{destination.name}.") and source.name.endswith(".tmp")
+            )
+            if not retryable:
+                raise
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
                 raise
-            time.sleep(0.005 * (attempt + 1))
+            delay = min(_WINDOWS_REPLACE_INITIAL_DELAY_SECONDS * (attempt + 1), remaining)
+            time.sleep(delay)
+            attempt += 1

tests/test_platform_edge_paths.py

@@ -40,19 +40,67 @@ def test_windows_replace_retries_transient_sharing_failure(self) -> None:
             source = Path(tmpdir) / "source"
             destination = Path(tmpdir) / "destination"
             source.write_text("payload", encoding="utf-8")
+            transient = PermissionError("busy")
+            transient.winerror = 32
             with (
                 mock.patch.object(private_files.os, "name", "nt"),
                 mock.patch.object(
                     private_files.os,
                     "replace",
-                    side_effect=[PermissionError("busy"), lambda src, dst: Path(dst).write_text(Path(src).read_text())],
+                    side_effect=[transient, lambda src, dst: Path(dst).write_text(Path(src).read_text())],
                 ) as replace,
                 mock.patch.object(private_files.time, "sleep") as sleep,
             ):
                 private_files._replace_with_retry(source, destination)  # pylint: disable=protected-access
             self.assertEqual(replace.call_count, 2)
             sleep.assert_called_once()
 
+    def test_windows_replace_fails_immediately_for_access_denied(self) -> None:
+        source = Path("source")
+        destination = Path("destination")
+        denied = PermissionError("access denied")
+        denied.winerror = 5
+        with (
+            mock.patch.object(private_files.os, "name", "nt"),
+            mock.patch.object(private_files.os, "replace", side_effect=denied),
+            mock.patch.object(private_files.time, "sleep") as sleep,
+        ):
+            with self.assertRaises(PermissionError) as raised:
+                private_files._replace_with_retry(source, destination)  # pylint: disable=protected-access
+        self.assertIs(raised.exception, denied)
+        sleep.assert_not_called()
+
+    def test_windows_replace_retries_compatibility_access_denied_for_owned_temp(self) -> None:
+        source = Path(".destination.owned.tmp")
+        destination = Path("destination")
+        transient = PermissionError("destination in use")
+        transient.winerror = 5
+        with (
+            mock.patch.object(private_files.os, "name", "nt"),
+            mock.patch.object(private_files.os, "replace", side_effect=[transient, None]) as replace,
+            mock.patch.object(private_files.time, "sleep") as sleep,
+        ):
+            private_files._replace_with_retry(source, destination)  # pylint: disable=protected-access
+        self.assertEqual(replace.call_count, 2)
+        sleep.assert_called_once()
+
+    def test_windows_replace_respects_elapsed_retry_deadline(self) -> None:
+        source = Path("source")
+        destination = Path("destination")
+        transient = PermissionError("busy")
+        transient.winerror = 33
+        clock = iter((0.0, 0.1, 1.0))
+        with (
+            mock.patch.object(private_files.os, "name", "nt"),
+            mock.patch.object(private_files.os, "replace", side_effect=transient),
+            mock.patch.object(private_files.time, "monotonic", side_effect=lambda: next(clock)),
+            mock.patch.object(private_files.time, "sleep") as sleep,
+        ):
+            with self.assertRaises(PermissionError) as raised:
+                private_files._replace_with_retry(source, destination)  # pylint: disable=protected-access
+        self.assertIs(raised.exception, transient)
+        self.assertEqual(sleep.call_count, 1)
+
     def test_parent_directory_open_is_disabled_on_windows(self) -> None:
         path = Path("/tmp")
         with mock.patch.object(private_files.os, "name", "nt"):