Treat Windows os.replace PermissionError as non-fatal cache miss

On Windows, os.replace raises PermissionError when another process
holds the target open. Instead of retrying, silently drop the write —
correct semantics for a cache (a missed write isn't corruption).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: cpcloud

cuda_core/cuda/core/utils/_program_cache.py

@@ -575,16 +575,14 @@ def __setitem__(self, key: object, value: object) -> None:
                 fh.write(record)
                 fh.flush()
                 os.fsync(fh.fileno())
-            # On Windows, os.replace can raise PermissionError when another
-            # process holds the target open.  Retry a few times with backoff.
-            for attempt in range(5):
-                try:
-                    os.replace(tmp_path, target)
-                    break
-                except PermissionError:
-                    if attempt == 4:
-                        raise
-                    time.sleep(0.01 * (2**attempt))
+            os.replace(tmp_path, target)
+        except PermissionError:
+            # On Windows, os.replace raises PermissionError when another
+            # process holds the target open. Treat as a non-fatal cache
+            # write failure — the entry simply won't be cached this time.
+            with contextlib.suppress(FileNotFoundError):
+                tmp_path.unlink()
+            return
         except BaseException:
             with contextlib.suppress(FileNotFoundError):
                 tmp_path.unlink()

cuda_core/tests/test_program_cache_multiprocess.py

@@ -61,7 +61,10 @@ def test_concurrent_writers_same_key_no_corruption(tmp_path):
         assert p.exitcode == 0, f"worker exited with {p.exitcode}"
 
     with FileStreamProgramCache(root) as cache:
-        got = cache[b"shared"]  # must not raise; payload is one of the writers'
+        # At least one writer must have succeeded; on Windows some writes
+        # may silently fail due to PermissionError on os.replace.
+        got = cache.get(b"shared")
+        assert got is not None, "no writer succeeded"
         assert bytes(got.code).startswith(b"v")