linkfile: Handle directory-to-symlink transitions safely

When a manifest changes from individual linkfiles inside a directory
(e.g. dest=".llms/rules", dest=".llms/skills") to a single linkfile
for the whole directory (e.g. dest=".llms", src="dot-llms"), two
things need to happen:

1. __linkIt must replace a real directory with a symlink.  Use
   os.rmdir() instead of platform_utils.remove() for real directories.
   rmdir only removes empty directories, so user-created content is
   never deleted.

2. UpdateCopyLinkfileList must handle the cleanup correctly:
   - Use os.rmdir() for directories (safe for non-empty)
   - Remove empty parent directories after cleaning old dests
   - Retry _CopyAndLinkFiles for all projects, since in interleaved
     sync mode _CopyAndLinkFiles runs before cleanup and may have
     failed because the directory was not yet empty

Change-Id: I0437b80beab98bce064cea81c11c47d699be91aa
Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/569243
Tested-by: Carlos Fernandez <carlosfsanz@meta.com>
Reviewed-by: Mike Frysinger <vapier@google.com>
Reviewed-by: Gavin Mak <gavinmak@google.com>
Commit-Queue: Carlos Fernandez <carlosfsanz@meta.com>

platform_utils.py

@@ -222,6 +222,43 @@ def rmdir(path):
     os.rmdir(_makelongpath(path))
 
 
+def removedirs(path: str) -> None:
+    """Remove a directory tree of symlinks and empty directories.
+
+    Walks |path| bottom-up without following symlinks.  Removes symlinks
+    and empty directories.  Stops at (and preserves) regular files and
+    non-empty directories.  Silently succeeds if |path| does not exist.
+
+    Availability: Unix, Windows.
+    """
+    if islink(path):
+        remove(path)
+        return
+
+    if not isdir(path):
+        return
+
+    for dirpath, dirnames, filenames in walk(path, topdown=False):
+        for name in dirnames:
+            entry = os.path.join(dirpath, name)
+            if islink(entry):
+                remove(entry)
+            else:
+                try:
+                    rmdir(entry)
+                except OSError:
+                    pass
+        for name in filenames:
+            entry = os.path.join(dirpath, name)
+            if islink(entry):
+                remove(entry)
+
+    try:
+        rmdir(path)
+    except OSError:
+        pass
+
+
 def isdir(path):
     """os.path.isdir(path) wrapper with support for long paths on Windows.
 

project.py

@@ -453,9 +453,13 @@ class _LinkFile(NamedTuple):
             platform_utils.readlink(absDest) != relSrc
         ):
             try:
-                # Remove existing file first, since it might be read-only.
+                # Remove existing path first, since it might be read-only.
                 if os.path.lexists(absDest):
-                    platform_utils.remove(absDest)
+                    # removedirs handles symlinks, empty dirs, and nested
+                    # trees of stale linkfile dests without deleting user
+                    # content.  Falls through to symlink() if the path was
+                    # fully removed, or raises OSError if not.
+                    platform_utils.removedirs(absDest)
                 else:
                     dest_dir = os.path.dirname(absDest)
                     if not platform_utils.isdir(dest_dir):

subcmds/sync.py

@@ -1631,9 +1631,10 @@ later is required to fix a server side protocol bug.
         new_paths = {}
         new_linkfile_paths = []
         new_copyfile_paths = []
-        for project in self.GetProjects(
+        projects = self.GetProjects(
             None, missing_ok=True, manifest=manifest, all_manifests=False
-        ):
+        )
+        for project in projects:
             new_linkfile_paths.extend(x.dest for x in project.linkfiles)
             new_copyfile_paths.extend(x.dest for x in project.copyfiles)
 
@@ -1669,17 +1670,36 @@ later is required to fix a server side protocol bug.
             )
 
             for need_remove_file in need_remove_files:
-                # Try to remove the updated copyfile or linkfile.
-                # So, if the file is not exist, nothing need to do.
-                platform_utils.remove(
-                    os.path.join(self.client.topdir, need_remove_file),
-                    missing_ok=True,
+                need_remove_path = os.path.join(
+                    self.client.topdir, need_remove_file
                 )
+                if os.path.isfile(need_remove_path):
+                    platform_utils.remove(need_remove_path)
+                else:
+                    platform_utils.removedirs(need_remove_path)
+
+                # Also try to remove empty parent directories.
+                parent = os.path.dirname(need_remove_path)
+                while parent != self.client.topdir:
+                    try:
+                        os.rmdir(parent)
+                    except OSError:
+                        break
+                    parent = os.path.dirname(parent)
 
         # Create copy-link-files.json, save dest path of "copyfile" and
         # "linkfile".
         with open(copylinkfile_path, "w", encoding="utf-8") as fp:
             json.dump(new_paths, fp)
+
+        # Retry linkfile/copyfile creation for all projects.  In
+        # interleaved sync mode, _CopyAndLinkFiles runs before this
+        # cleanup, so linkfiles whose dest was blocked by an old
+        # directory may have failed.  _CopyAndLinkFiles is idempotent
+        # and skips dests that are already correct.
+        for project in projects:
+            project._CopyAndLinkFiles()
+
         return True
 
     def _SmartSyncSetup(self, opt, smart_sync_manifest_path, manifest):

tests/test_platform_utils.py

@@ -46,3 +46,71 @@ def test_remove_missing_ok(tmp_path: Path) -> None:
     path.touch()
     platform_utils.remove(path, missing_ok=False)
     assert not path.exists()
+
+
+def test_removedirs_nonexistent(tmp_path: Path) -> None:
+    """removedirs should silently succeed on nonexistent paths."""
+    platform_utils.removedirs(tmp_path / "does-not-exist")
+
+
+def test_removedirs_symlink(tmp_path: Path) -> None:
+    """removedirs should remove a symlink."""
+    link = tmp_path / "link"
+    link.symlink_to("target")
+    platform_utils.removedirs(link)
+    assert not link.exists()
+
+
+def test_removedirs_empty_dir(tmp_path: Path) -> None:
+    """removedirs should remove an empty directory."""
+    d = tmp_path / "empty"
+    d.mkdir()
+    platform_utils.removedirs(d)
+    assert not d.exists()
+
+
+def test_removedirs_nested_empty_dirs(tmp_path: Path) -> None:
+    """removedirs should remove nested empty directories."""
+    d = tmp_path / "a" / "b" / "c"
+    d.mkdir(parents=True)
+    platform_utils.removedirs(tmp_path / "a")
+    assert not (tmp_path / "a").exists()
+
+
+def test_removedirs_symlinks_inside_dir(tmp_path: Path) -> None:
+    """removedirs should remove symlinks inside a directory."""
+    d = tmp_path / "dir"
+    d.mkdir()
+    (d / "link1").symlink_to("target1")
+    (d / "link2").symlink_to("target2")
+    platform_utils.removedirs(d)
+    assert not d.exists()
+
+
+def test_removedirs_preserves_user_files(tmp_path: Path) -> None:
+    """removedirs should not delete regular files or their parent dirs."""
+    d = tmp_path / "dir"
+    d.mkdir()
+    (d / "link").symlink_to("target")
+    (d / "user-file.txt").write_text("keep me")
+    platform_utils.removedirs(d)
+    assert d.exists()
+    assert not (d / "link").exists()
+    assert (d / "user-file.txt").read_text() == "keep me"
+
+
+def test_removedirs_deep_nested_with_symlinks(tmp_path: Path) -> None:
+    """removedirs should handle deep nesting: sub/dir/target."""
+    d = tmp_path / "sub" / "dir"
+    d.mkdir(parents=True)
+    (d / "link").symlink_to("target")
+    platform_utils.removedirs(tmp_path / "sub")
+    assert not (tmp_path / "sub").exists()
+
+
+def test_removedirs_regular_file_noop(tmp_path: Path) -> None:
+    """removedirs should not delete a regular file."""
+    f = tmp_path / "file.txt"
+    f.write_text("data")
+    platform_utils.removedirs(f)
+    assert f.exists()

tests/test_project.py

@@ -365,6 +365,47 @@ class LinkFile(CopyLinkTestCase):
             os.path.join("git-project", "foo.txt"), os.readlink(dest)
         )
 
+    def test_replace_empty_dir_with_symlink(self):
+        """A linkfile should replace an empty real directory at the dest path.
+
+        This is the common case: the old linkfiles inside the directory were
+        already cleaned up by UpdateCopyLinkfileList, leaving an empty parent
+        directory behind.
+        """
+        src_dir = os.path.join(self.worktree, "dot-llms")
+        os.makedirs(src_dir)
+
+        dest = os.path.join(self.topdir, "mydir")
+        os.makedirs(dest)
+
+        lf = self.LinkFile("dot-llms", "mydir")
+        lf._Link()
+        self.assertTrue(os.path.islink(dest))
+        self.assertEqual(
+            os.path.join("git-project", "dot-llms"), os.readlink(dest)
+        )
+
+    def test_nonempty_dir_not_clobbered(self):
+        """A linkfile must not delete a non-empty directory.
+
+        If the user created files in a directory that a new linkfile wants
+        to replace, __linkIt should fail safely rather than deleting content.
+        """
+        src_dir = os.path.join(self.worktree, "dot-llms")
+        os.makedirs(src_dir)
+
+        dest = os.path.join(self.topdir, "mydir")
+        os.makedirs(dest)
+        user_file = os.path.join(dest, "user-notes.txt")
+        self.touch(user_file)
+
+        lf = self.LinkFile("dot-llms", "mydir")
+        lf._Link()
+        # The directory should NOT be replaced — user content is preserved.
+        self.assertFalse(os.path.islink(dest))
+        self.assertTrue(os.path.isdir(dest))
+        self.assertTrue(os.path.exists(user_file))
+
 
 class MigrateWorkTreeTests(unittest.TestCase):
     """Check _MigrateOldWorkTreeGitDir handling."""

tests/test_subcmds_sync.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 """Unittests for the subcmds/sync.py module."""
 
+import json
 import os
 import shutil
 import tempfile
@@ -1145,3 +1146,143 @@ class InterleavedSyncTest(unittest.TestCase):
         self.assertTrue(result.checkout_success)
         project.Sync_NetworkHalf.assert_called_once()
         project.Sync_LocalHalf.assert_not_called()
+
+
+class UpdateCopyLinkfileListTest(unittest.TestCase):
+    """Tests for Sync.UpdateCopyLinkfileList."""
+
+    def setUp(self):
+        self.tempdirobj = tempfile.TemporaryDirectory(prefix="repo_tests")
+        self.topdir = self.tempdirobj.name
+        self.repodir = os.path.join(self.topdir, ".repo")
+        os.makedirs(self.repodir)
+
+        manifest = mock.MagicMock()
+        manifest.subdir = self.repodir
+        self.manifest = manifest
+
+        git_event_log = mock.MagicMock(ErrorEvent=mock.Mock(return_value=None))
+        self.cmd = sync.Sync(
+            manifest=manifest,
+            outer_client=mock.MagicMock(),
+            git_event_log=git_event_log,
+        )
+        self.cmd.client = mock.MagicMock(topdir=self.topdir)
+
+    def tearDown(self):
+        self.tempdirobj.cleanup()
+
+    def _write_copylinkfile_json(self, data: dict) -> None:
+        path = os.path.join(self.repodir, "copy-link-files.json")
+        with open(path, "w") as f:
+            json.dump(data, f)
+
+    def _setup_projects(self, linkfile_dests: list) -> None:
+        project = mock.MagicMock()
+        project.linkfiles = [mock.MagicMock(dest=d) for d in linkfile_dests]
+        project.copyfiles = []
+        mock.patch.object(
+            self.cmd, "GetProjects", return_value=[project]
+        ).start()
+
+    def test_removes_old_symlink_dest(self):
+        """Old linkfile dests that are symlinks should be removed."""
+        old_dest = os.path.join(self.topdir, "old-link")
+        os.symlink("target", old_dest)
+
+        self._write_copylinkfile_json(
+            {"linkfile": ["old-link"], "copyfile": []}
+        )
+        self._setup_projects([])
+
+        self.cmd.UpdateCopyLinkfileList(self.manifest)
+        self.assertFalse(os.path.lexists(old_dest))
+
+    def test_does_not_delete_through_new_symlink(self):
+        """Old dests that resolve through a new symlink must not delete files.
+
+        When the manifest changes from individual linkfiles inside a directory
+        to a single directory linkfile, and _CopyAndLinkFiles has already
+        created the symlink (interleaved mode), cleanup must not follow the
+        symlink and delete real project files.
+        """
+        project_dir = os.path.join(self.topdir, "vendor", "tools", "llms")
+        os.makedirs(os.path.join(project_dir, "dot-llms", "rules"))
+        os.makedirs(os.path.join(project_dir, "dot-llms", "skills"))
+        with open(
+            os.path.join(project_dir, "dot-llms", "rules", "basics.md"), "w"
+        ) as f:
+            f.write("# basics")
+        with open(
+            os.path.join(project_dir, "dot-llms", "skills", "repo.md"), "w"
+        ) as f:
+            f.write("# repo")
+
+        # Simulate interleaved mode: .llms -> vendor/tools/llms/dot-llms.
+        llms_link = os.path.join(self.topdir, ".llms")
+        os.symlink("vendor/tools/llms/dot-llms", llms_link)
+
+        self._write_copylinkfile_json(
+            {"linkfile": [".llms/rules", ".llms/skills"], "copyfile": []}
+        )
+        self._setup_projects([".llms"])
+
+        self.cmd.UpdateCopyLinkfileList(self.manifest)
+
+        # Real project files must still exist.
+        self.assertTrue(
+            os.path.exists(
+                os.path.join(project_dir, "dot-llms", "rules", "basics.md")
+            )
+        )
+        self.assertTrue(
+            os.path.exists(
+                os.path.join(project_dir, "dot-llms", "skills", "repo.md")
+            ),
+        )
+        self.assertTrue(os.path.islink(llms_link))
+
+    def test_cleans_up_empty_parent_dirs(self):
+        """After removing old dests, empty parent directories are removed."""
+        llms_dir = os.path.join(self.topdir, ".llms")
+        os.makedirs(llms_dir)
+        os.symlink(
+            "../vendor/tools/llms/rules", os.path.join(llms_dir, "rules")
+        )
+        os.symlink(
+            "../vendor/tools/llms/skills",
+            os.path.join(llms_dir, "skills"),
+        )
+
+        self._write_copylinkfile_json(
+            {"linkfile": [".llms/rules", ".llms/skills"], "copyfile": []}
+        )
+        self._setup_projects([".llms"])
+
+        self.cmd.UpdateCopyLinkfileList(self.manifest)
+
+        self.assertFalse(os.path.lexists(os.path.join(llms_dir, "rules")))
+        self.assertFalse(os.path.lexists(os.path.join(llms_dir, "skills")))
+        # Parent directory should be removed since it's now empty.
+        self.assertFalse(os.path.exists(llms_dir))
+
+    def test_preserves_nonempty_parent_dirs(self):
+        """Non-empty parent directories are preserved after old dest removal."""
+        llms_dir = os.path.join(self.topdir, ".llms")
+        os.makedirs(llms_dir)
+        os.symlink(
+            "../vendor/tools/llms/rules", os.path.join(llms_dir, "rules")
+        )
+        with open(os.path.join(llms_dir, "my-notes.txt"), "w") as f:
+            f.write("user content")
+
+        self._write_copylinkfile_json(
+            {"linkfile": [".llms/rules"], "copyfile": []}
+        )
+        self._setup_projects([".llms"])
+
+        self.cmd.UpdateCopyLinkfileList(self.manifest)
+
+        self.assertFalse(os.path.lexists(os.path.join(llms_dir, "rules")))
+        self.assertTrue(os.path.exists(os.path.join(llms_dir, "my-notes.txt")))
+        self.assertTrue(os.path.isdir(llms_dir))