refactor(paths): use direct lstat probing in collision resolution

media_shrinker.py

@@ -11,6 +11,7 @@
 import functools
 from concurrent.futures import ThreadPoolExecutor, as_completed
 import json
+import errno
 import math
 import os
 import re
@@ -2201,12 +2202,30 @@ def _ensure_not_source_path(source: Path, output: Path) -> None:
 
 def _resolve_collision(path: Path, *, overwrite: bool) -> Path:
     """Return path or a numbered variant if path already exists."""
-    if overwrite or not path.exists():
+    if overwrite:
         return path
+
+    # Invariant: Only ENOENT proves a candidate is completely free.
+    # Dangling symlinks or files with access errors remain occupied (fail-closed).
+    try:
+        os.lstat(path)
+    except OSError as exc:
+        if exc.errno == errno.ENOENT:
+            return path
+
+    parent_str = str(path.parent)
+    stem_str = path.stem
+    suffix_str = path.suffix
+
     for index in range(1, 10_000):
-        candidate = path.with_name(f"{path.stem}-{index}{path.suffix}")
-        if not candidate.exists():
-            return candidate
+        candidate_str = os.path.join(parent_str, f"{stem_str}-{index}{suffix_str}")
+        try:
+            os.lstat(candidate_str)
+        except OSError as exc:
+            if exc.errno == errno.ENOENT:
+                return Path(candidate_str)
+            continue
+
     raise FileExistsError(f"Could not find free output path for {path}")
 
 

tests/test_media_shrinker.py

@@ -227,7 +227,6 @@ def test_find_candidates_skips_entries_when_symlink_check_fails(self) -> None:
             nested.write_bytes(b"0" * 4)
 
             import os
-
             original_lstat = os.lstat
 
             def flaky_lstat(path):
@@ -1772,6 +1771,65 @@ def test_resolve_collision_returns_original_if_overwrite_is_true(self) -> None:
             resolved = media_shrinker._resolve_collision(path, overwrite=True)
             self.assertEqual(resolved, path)
 
+    def test_resolve_collision_handles_dangling_symlinks_on_original_path(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "existing.flac"
+            path.symlink_to("does_not_exist.flac")
+            resolved = media_shrinker._resolve_collision(path, overwrite=False)
+            self.assertEqual(resolved, Path(tmp) / "existing-1.flac")
+
+    def test_resolve_collision_handles_permission_errors_on_original_path(self) -> None:
+        import errno
+        original_lstat = os.lstat
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "existing.flac"
+
+            def side_effect(candidate_str, *args, **kwargs):
+                if str(candidate_str).endswith("existing.flac"):
+                    e = OSError()
+                    e.errno = errno.EACCES
+                    raise e
+                elif "-1.flac" in str(candidate_str):
+                    e = OSError()
+                    e.errno = errno.ENOENT
+                    raise e
+                return original_lstat(candidate_str, *args, **kwargs)
+
+            with patch("os.lstat", side_effect=side_effect):
+                resolved = media_shrinker._resolve_collision(path, overwrite=False)
+                self.assertEqual(resolved, Path(tmp) / "existing-1.flac")
+
+    def test_resolve_collision_handles_dangling_symlinks(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "existing.flac"
+            path.write_bytes(b"data")
+            dangling = Path(tmp) / "existing-1.flac"
+            dangling.symlink_to("does_not_exist.flac")
+            resolved = media_shrinker._resolve_collision(path, overwrite=False)
+            self.assertEqual(resolved, Path(tmp) / "existing-2.flac")
+
+    def test_resolve_collision_handles_permission_errors(self) -> None:
+        import errno
+        original_lstat = os.lstat
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "existing.flac"
+            path.write_bytes(b"data")
+
+            def side_effect(candidate_str, *args, **kwargs):
+                if isinstance(candidate_str, str) and "-1.flac" in candidate_str:
+                    e = OSError()
+                    e.errno = errno.EACCES
+                    raise e
+                elif isinstance(candidate_str, str) and "-2.flac" in candidate_str:
+                    e = OSError()
+                    e.errno = errno.ENOENT
+                    raise e
+                return original_lstat(candidate_str, *args, **kwargs)
+
+            with patch("os.lstat", side_effect=side_effect):
+                resolved = media_shrinker._resolve_collision(path, overwrite=False)
+                self.assertEqual(resolved, Path(tmp) / "existing-2.flac")
+
 
 class CliTests(unittest.TestCase):
     def test_normalize_argv_handles_silence_noise_values(self) -> None:
@@ -2433,7 +2491,7 @@ def test_resolve_collision_reports_exhausted_names(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             path = Path(tmp) / "existing.flac"
             path.write_bytes(b"data")
-            with patch("pathlib.Path.exists", return_value=True):
+            with patch("os.lstat", return_value=True), patch("pathlib.Path.exists", return_value=True):
                 with self.assertRaisesRegex(FileExistsError, "Could not find free"):
                     media_shrinker._resolve_collision(path, overwrite=False)