fix(presets): keep user files when removing preset skills

src/specify_cli/integrations/_file_changes.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import errno
+import os
+import stat
 from collections.abc import Callable
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
 from contextvars import ContextVar
 from pathlib import Path
 
@@ -45,3 +48,111 @@ def write_bytes(path: Path, content: bytes) -> int:
 def unlink(path: Path, *, missing_ok: bool = False) -> None:
     with changing_file(path, removal=True):
         path.unlink(missing_ok=missing_ok)
+
+
+def unlink_in_directory(path: Path, *, root: Path) -> None:
+    """Remove a generated file and its empty parent beneath a trusted root."""
+    directory = path.absolute().parent
+    root = root.absolute()
+    try:
+        relative = directory.relative_to(root)
+    except ValueError:
+        return
+    if not relative.parts or ".." in relative.parts:
+        return
+    if (
+        getattr(os, "O_DIRECTORY", 0)
+        and getattr(os, "O_NOFOLLOW", 0)
+        and {os.open, os.unlink, os.rmdir, os.stat} <= os.supports_dir_fd
+        and os.listdir in os.supports_fd
+    ):
+        flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
+        with ExitStack() as stack:
+            try:
+                parent_fd = os.open(root, flags & ~os.O_NOFOLLOW)
+                stack.callback(os.close, parent_fd)
+                for part in relative.parts[:-1]:
+                    parent_fd = os.open(part, flags, dir_fd=parent_fd)
+                    stack.callback(os.close, parent_fd)
+                directory_fd = os.open(directory.name, flags, dir_fd=parent_fd)
+                stack.callback(os.close, directory_fd)
+            except OSError as exc:
+                if exc.errno not in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR):
+                    raise
+                return
+            with changing_file(path, removal=True):
+                try:
+                    os.unlink(path.name, dir_fd=directory_fd)
+                except FileNotFoundError:
+                    pass
+
+            def same_directory() -> bool:
+                return os.path.samestat(
+                    os.fstat(directory_fd),
+                    os.stat(directory.name, dir_fd=parent_fd, follow_symlinks=False),
+                )
+
+            try:
+                empty = not os.listdir(directory_fd) and same_directory()
+            except (FileNotFoundError, NotADirectoryError):
+                return
+            if empty:
+                before_file_change(path.parent, removal=True)
+                try:
+                    if not same_directory():
+                        return
+                    os.rmdir(directory.name, dir_fd=parent_fd)
+                except OSError as exc:
+                    if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST, errno.ENOENT, errno.ENOTDIR):
+                        raise
+                    return
+                after_file_change(path.parent)
+        return
+
+    # Without dir_fd (notably Windows), re-check before each path-based
+    # deletion. This narrows, but cannot eliminate, concurrent swap races.
+    def safe_directory() -> os.stat_result | None:
+        parent = root
+        for part in relative.parts:
+            parent = parent / part
+            try:
+                info = parent.lstat()
+            except OSError as exc:
+                if exc.errno not in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR):
+                    raise
+                return None
+            if not stat.S_ISDIR(info.st_mode) or (
+                getattr(info, "st_file_attributes", 0)
+                & stat.FILE_ATTRIBUTE_REPARSE_POINT
+            ):
+                return None
+        return info
+
+    original = safe_directory()
+    if original is None:
+        return
+    with changing_file(path, removal=True):
+        if safe_directory() is None:
+            return
+        try:
+            path.unlink()
+        except FileNotFoundError:
+            pass
+    try:
+        empty = not any(directory.iterdir())
+    except OSError as exc:
+        if exc.errno not in (errno.ENOENT, errno.ELOOP, errno.ENOTDIR):
+            raise
+        return
+    if empty:
+        before_file_change(path.parent, removal=True)
+        current = safe_directory()
+        if current is None or not os.path.samestat(original, current):
+            return
+        try:
+            directory.rmdir()
+        except OSError as exc:
+            if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST, errno.ENOENT, errno.ENOTDIR):
+                raise
+            return
+        after_file_change(path.parent)

src/specify_cli/presets/_manager_skills.py

@@ -2,7 +2,6 @@
 
 import copy
 import os
-import shutil
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
 
@@ -216,7 +215,7 @@ def apply_to_dir(
                 managed_skill_names if managed_names is None else managed_names
             )
             # Restore skills for commands whose winner is non-preset.
-            # _unregister_skills_in_dir can rmtree the skill dir, so
+            # _unregister_skills_in_dir can remove the skill file, so
             # overrides must be handled directly (create dir + write)
             # without that call.
             dir_core_ext_names = [
@@ -1275,11 +1274,12 @@ def _delete_agent_preset_skills(
             if override_source:
                 owned_sources.add(override_source)
             if source in owned_sources:
-                from ..integrations._file_changes import after_file_change, before_file_change
+                from ..integrations._file_changes import unlink_in_directory
 
-                before_file_change(skill_subdir, removal=True)
-                shutil.rmtree(skill_subdir)
-                after_file_change(skill_subdir)
+                unlink_in_directory(
+                    skill_file,
+                    root=self.project_root if skill_file.is_relative_to(self.project_root) else skills_dir,
+                )
 
     @staticmethod
     def _warn_unrestored_skill(
@@ -1370,7 +1370,7 @@ def _unregister_skills_in_dir(
             # is_dir() follows symlinks, so a symlinked skill subdirectory
             # (with a safe, non-symlinked parent) would otherwise slip past
             # _safe_skills_dir_for_agent's parent-only check and have
-            # write_text/rmtree operate through it (#2948).
+            # writes/removals operate through it (#2948).
             if not self._validate_skill_subdir(
                 skill_subdir, create=False, skills_root=skills_dir
             ):
@@ -1435,7 +1435,7 @@ def _unregister_skills_in_dir(
                 # source cannot produce restored content, so leave the
                 # existing skill untouched rather than leaking a raw
                 # OSError/UnicodeDecodeError out of `preset remove` — and
-                # rather than falling through to the rmtree below, which
+                # rather than falling through to the removal below, which
                 # would delete a skill precisely when its replacement
                 # cannot be generated. Matches the `continue` guards above
                 # (unsafe name, missing subdir, foreign owner), which also
@@ -1541,12 +1541,13 @@ def _unregister_skills_in_dir(
                 _write_shared_text(skills_dir, skill_file, skill_content)
                 mutated_names.append(skill_name)
             else:
-                # No core or extension template — remove the skill entirely
-                from ..integrations._file_changes import after_file_change, before_file_change
+                # No core or extension template — remove only the generated file
+                from ..integrations._file_changes import unlink_in_directory
 
-                before_file_change(skill_subdir, removal=True)
-                shutil.rmtree(skill_subdir)
-                after_file_change(skill_subdir)
+                unlink_in_directory(
+                    skill_file,
+                    root=self.project_root if skill_file.is_relative_to(self.project_root) else skills_dir,
+                )
                 mutated_names.append(skill_name)
 
         return mutated_names

tests/specify_cli/integrations/test_command_switch.py

@@ -25,6 +25,12 @@
     _write_invalid_manifest,  # noqa: F401
     runner,  # noqa: F401
 )
+from tests.specify_cli.presets._skill_cleanup_helpers import (
+    INTEGRATION_SWITCH,
+    check_cleanup_preserves_swapped_link_target,
+    check_cleanup_preserves_user_files,
+    check_cleanup_rollback_preserves_directory_mode,
+)
 
 class TestIntegrationSwitch:
     @pytest.mark.parametrize("installed", [False, True])
@@ -1188,3 +1194,29 @@ def fail_cleanup(self, integration_key):
         assert "Failed to clean up extension artifacts for integration 'copilot'" in normalized
         assert "cleanup exploded" in normalized
         assert "Switched to integration" in normalized
+
+
+@pytest.mark.parametrize("fallback", [False, True])
+@pytest.mark.parametrize("ancestor", [False, True])
+def test_switch_preset_skill_cleanup_preserves_swapped_link_target(tmp_path, monkeypatch, fallback, ancestor):
+    check_cleanup_preserves_swapped_link_target(tmp_path, monkeypatch, INTEGRATION_SWITCH, fallback, ancestor)
+
+
+@pytest.mark.parametrize(
+    "user_reference,layout",
+    [(False, "project"), (True, "project"), (False, "linked-parent")],
+    ids=["empty", "user-files", "linked-parent"],
+)
+@pytest.mark.parametrize("fallback", [False, True])
+def test_switch_preset_skill_cleanup_preserves_user_files(tmp_path, monkeypatch, user_reference, layout, fallback):
+    check_cleanup_preserves_user_files(tmp_path, monkeypatch, INTEGRATION_SWITCH, user_reference, layout, fallback)
+
+
+@pytest.mark.parametrize("user_reference", [False, True])
+@pytest.mark.parametrize("fallback", [False, True])
+def test_switch_preset_skill_cleanup_rollback_preserves_directory_mode(
+    tmp_path, monkeypatch, capsys, user_reference, fallback,
+):
+    check_cleanup_rollback_preserves_directory_mode(
+        tmp_path, monkeypatch, capsys, INTEGRATION_SWITCH, user_reference, fallback
+    )