fix(utils): retry transient atomic replace permission errors

tests/hermes_cli/test_atomic_json_write.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from utils import atomic_json_write
+from utils import atomic_json_write, atomic_replace
 
 
 class TestAtomicJsonWrite:
@@ -165,6 +165,28 @@ def test_mode_applied_when_supported(self, tmp_path):
             actual = stat_mod.S_IMODE(target.stat().st_mode)
             assert actual == 0o600
 
+    def test_atomic_replace_retries_transient_permission_error(self, tmp_path):
+        import utils
+
+        target = tmp_path / "data.json"
+        tmp_file = tmp_path / "data.tmp"
+        tmp_file.write_text('{"ok": true}', encoding="utf-8")
+        calls = 0
+        real_replace = os.replace
+
+        def flaky_replace(src, dst):
+            nonlocal calls
+            calls += 1
+            if calls < 3:
+                raise PermissionError("temporarily locked")
+            real_replace(src, dst)
+
+        with patch.object(utils.os, "replace", side_effect=flaky_replace):
+            assert atomic_replace(tmp_file, target) == str(target)
+
+        assert calls == 3
+        assert json.loads(target.read_text(encoding="utf-8")) == {"ok": True}
+
     def test_concurrent_writes_dont_corrupt(self, tmp_path):
         """Multiple rapid writes should each produce valid JSON."""
         import threading

utils.py

@@ -5,6 +5,7 @@
 import os
 import stat
 import tempfile
+import time
 from pathlib import Path
 from typing import Any, Union
 from urllib.parse import urlparse
@@ -15,6 +16,8 @@
 
 
 TRUTHY_STRINGS = frozenset({"1", "true", "yes", "on"})
+_ATOMIC_REPLACE_MAX_ATTEMPTS = 8
+_ATOMIC_REPLACE_RETRY_DELAY = 0.01
 
 
 def is_truthy_value(value: Any, default: bool = False) -> bool:
@@ -78,7 +81,14 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
     """
     target_str = str(target)
     real_path = os.path.realpath(target_str) if os.path.islink(target_str) else target_str
-    os.replace(str(tmp_path), real_path)
+    for attempt in range(_ATOMIC_REPLACE_MAX_ATTEMPTS):
+        try:
+            os.replace(str(tmp_path), real_path)
+            break
+        except PermissionError:
+            if attempt == _ATOMIC_REPLACE_MAX_ATTEMPTS - 1:
+                raise
+            time.sleep(_ATOMIC_REPLACE_RETRY_DELAY * (attempt + 1))
     return real_path