fix: make local registry concurrent delete parent cleanup race-safe

mindtrace/registry/mindtrace/registry/backends/local_registry_backend.py

@@ -573,14 +573,19 @@ def delete(
                     if meta_path.exists():
                         meta_path.unlink()
 
-                    # Cleanup parent if empty
+                    # Cleanup parent if empty (race-safe under concurrent deletes).
+                    # Another thread/process may remove `parent` between checks/iteration.
                     parent = target.parent
-                    if parent.exists() and not any(parent.iterdir()):
-                        self.logger.debug(f"Removing empty parent directory: {parent}")
-                        try:
-                            parent.rmdir()
-                        except Exception:
-                            pass
+                    try:
+                        if not any(parent.iterdir()):
+                            self.logger.debug(f"Removing empty parent directory: {parent}")
+                            try:
+                                parent.rmdir()
+                            except Exception:
+                                pass
+                    except FileNotFoundError:
+                        # Parent already removed by a concurrent delete; this is fine.
+                        pass
 
                 results.add(OpResult.success(obj_name, obj_version))
             except Exception as e:

tests/unit/mindtrace/registry/backends/test_local_registry_backend.py

@@ -304,6 +304,25 @@ def mock_rmdir(self):
         assert not object_path.exists()
 
 
+def test_delete_parent_cleanup_race_on_iterdir(backend, sample_object_dir, sample_metadata):
+    """Delete should succeed when parent disappears before parent.iterdir() runs."""
+    backend.push("test:race", "1.0.0", str(sample_object_dir), sample_metadata)
+
+    parent_path = backend.uri / "test:race"
+    original_iterdir = Path.iterdir
+
+    def mock_iterdir(self):
+        if self == parent_path:
+            raise FileNotFoundError("Simulated concurrent parent removal")
+        return original_iterdir(self)
+
+    with patch.object(Path, "iterdir", mock_iterdir):
+        result = backend.delete("test:race", "1.0.0")
+
+    assert result[("test:race", "1.0.0")].ok
+    assert not (backend.uri / "test:race" / "1.0.0").exists()
+
+
 def test_init_with_file_uri(temp_dir):
     """Test LocalRegistryBackend initialization with file:// URI."""
     # Test that file:// prefix is stripped