Retry transient Windows locks when publishing Tailwind

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+- Retry transient Windows access and sharing errors when publishing the
+  verified Tailwind executable, with a bounded wait and no repeated download.
+
 ## 0.3.0 — 2026-10-19
 
 - Discover Chrome/Chromium without launching it; optionally provision pinned,

sdkmgr/tailwind.py

@@ -5,6 +5,7 @@
 import subprocess
 import sys
 import tempfile
+import time
 from pathlib import Path
 
 from filelock import FileLock, Timeout
@@ -95,6 +96,20 @@ def find(root: Path, host: Platform, version: str, *, executable=None) -> ToolIn
         raise SDKError(f"Cannot inspect Tailwind installation: {error}") from error
 
 
+def _publish(staged: Path, directory: Path):
+    # Windows can briefly retain an executable lock after the version probe.
+    delays = iter((0.05, 0.1, 0.2, 0.4, 0.8, 1.6))
+    while True:
+        try:
+            os.replace(staged, directory)
+            return
+        except OSError as error:
+            delay = next(delays, None)
+            if getattr(error, "winerror", None) not in {5, 32} or delay is None:
+                raise
+            time.sleep(delay)
+
+
 def ensure(root: Path, host: Platform, version: str, *, executable=None) -> ToolInstallation:
     if existing := find(root, host, version, executable=executable):
         return existing
@@ -118,7 +133,7 @@ def ensure(root: Path, host: Platform, version: str, *, executable=None) -> Tool
                 binary.chmod(0o755)
                 if _version(binary) != version:
                     raise SDKError(f"Downloaded Tailwind did not report version {version}.")
-                os.replace(staged, directory)
+                _publish(staged, directory)
             result = _cached(directory, host, version, asset)
             assert result is not None
             return result

tests/test_tailwind.py

@@ -70,6 +70,58 @@ def find(*args, **kwargs):
     assert len(downloads) == 2
 
 
+@pytest.mark.parametrize("winerror", [5, 32])
+def test_publication_recovers_from_windows_lock(tmp_path, monkeypatch, downloads, winerror):
+    replace = tailwind.os.replace
+    attempts = []
+    sleeps = []
+
+    def locked_once(source, destination):
+        attempts.append((source, destination))
+        assert not destination.exists()
+        assert (source / "tailwindcss").read_bytes() == PAYLOAD
+        if len(attempts) == 1:
+            error = PermissionError("Executable is still locked")
+            error.winerror = winerror
+            raise error
+        replace(source, destination)
+
+    monkeypatch.setattr(tailwind.os, "replace", locked_once)
+    monkeypatch.setattr(tailwind.time, "sleep", sleeps.append)
+    tool = Manager(tmp_path).ensure("tailwindcss", version=VERSION)
+    assert tool.path.read_bytes() == PAYLOAD
+    assert len(attempts) == 2
+    assert attempts[0] == attempts[1]
+    assert sleeps == [0.05]
+    assert len(downloads) == 2
+    assert not list(tmp_path.rglob(".install-*"))
+
+
+@pytest.mark.parametrize("winerror,attempts", [(5, 7), (32, 7), (2, 1), (None, 1)])
+def test_publication_errors_are_bounded(tmp_path, monkeypatch, downloads, winerror, attempts):
+    failure = PermissionError("Cannot publish")
+    if winerror is not None:
+        failure.winerror = winerror
+    calls = []
+    sleeps = []
+
+    def fail(source, destination):
+        calls.append((source, destination))
+        raise failure
+
+    monkeypatch.setattr(tailwind.os, "replace", fail)
+    monkeypatch.setattr(tailwind.time, "sleep", sleeps.append)
+    with pytest.raises(SDKError, match="Could not install Tailwind: Cannot publish") as caught:
+        Manager(tmp_path).ensure("tailwindcss", version=VERSION)
+    assert caught.value.__cause__ is failure
+    assert len(calls) == attempts
+    assert len(sleeps) == attempts - 1
+    assert sum(sleeps) == pytest.approx(3.15 if attempts > 1 else 0)
+    assert not calls[0][1].exists()
+    assert len(downloads) == 2
+    assert not list(tmp_path.rglob(".install-*"))
+
+
 def test_lock_timeout_is_diagnostic(tmp_path, monkeypatch):
     class Locked:
         def __init__(self, path, **kwargs):