feat: add builtin uninstaller as alternative to calling pip

docs/configuration.md

@@ -307,6 +307,21 @@ values, usage instructions and warnings.
 
 Use parallel execution when using the new (`>=1.1.0`) installer.
 
+### `installer.builtin-uninstall`
+
+**Type**: `boolean`
+
+**Default**: `false`
+
+**Environment Variable**: `POETRY_INSTALLER_BUILTIN_UNINSTALL`
+
+*Introduced in 2.5.0*
+
+If set to `true`, Poetry uninstalls packages using its own built-in routine instead of
+invoking `pip uninstall` as a subprocess. This avoids the overhead of spawning pip.
+Behavior is otherwise equivalent: confirmation is automatic, and files outside the
+target environment's prefix are never removed.
+
 ### `installer.build-config-settings.<package-name>`
 
 **Type**: `Serialised JSON with string or list of string properties`

src/poetry/config/config.py

@@ -174,6 +174,7 @@ class Config:
             "no-binary": None,
             "only-binary": None,
             "build-config-settings": {},
+            "builtin-uninstall": False,
         },
         "python": {"installation-dir": os.path.join("{data-dir}", "python")},
         "solver": {
@@ -396,6 +397,7 @@ def _get_normalizer(name: str | Sequence[str]) -> Callable[[str], Any]:
             "virtualenvs.use-poetry-python",
             "installer.re-resolve",
             "installer.parallel",
+            "installer.builtin-uninstall",
             "solver.lazy-wheel",
             "system-git-client",
             "keyring.enabled",

src/poetry/console/commands/config.py

@@ -93,6 +93,7 @@ def unique_config_values(self) -> dict[str, tuple[Any, Any]]:
             "requests.max-retries": (lambda val: int(val) >= 0, int_normalizer),
             "installer.re-resolve": (boolean_validator, boolean_normalizer),
             "installer.parallel": (boolean_validator, boolean_normalizer),
+            "installer.builtin-uninstall": (boolean_validator, boolean_normalizer),
             "installer.max-workers": (lambda val: int(val) > 0, int_normalizer),
             "installer.no-binary": (
                 PackageFilterPolicy.validator,

src/poetry/installation/executor.py

@@ -21,6 +21,8 @@
 from poetry.installation.operations import Install
 from poetry.installation.operations import Uninstall
 from poetry.installation.operations import Update
+from poetry.installation.uninstaller import UninstallPathSet
+from poetry.installation.uninstaller import uninstall_distribution
 from poetry.installation.wheel_installer import WheelInstaller
 from poetry.puzzle.exceptions import SolverProblemError
 from poetry.utils._compat import decode
@@ -78,6 +80,9 @@ def __init__(
         self._verbose = False
         self._wheel_installer = WheelInstaller(self._env)
         self._build_constraints = build_constraints or {}
+        self._use_builtin_uninstall: bool = config.get(
+            "installer.builtin-uninstall", False
+        )
 
         if parallel is None:
             parallel = config.get("installer.parallel", True)
@@ -603,15 +608,27 @@ def _install(self, operation: Install | Update) -> int:
         )
         self._write(operation, message)
 
+        old_pathset: UninstallPathSet | None = None
         try:
             if operation.job_type == "update":
-                # Uninstall first
-                # TODO: Make an uninstaller and find a way to rollback in case
-                # the new package can't be installed
                 assert isinstance(operation, Update)
-                self._remove(operation.initial_package)
+                result = self._uninstall(operation.initial_package)
+                if isinstance(result, UninstallPathSet):
+                    old_pathset = result
+                elif result != 0:
+                    # pip uninstall was interrupted; surface that code
+                    # instead of attempting the install.
+                    return result
 
-            self._wheel_installer.install(archive)
+            try:
+                self._wheel_installer.install(archive)
+            except BaseException:  # rollback also on KeyboardInterrupt
+                if old_pathset is not None:
+                    old_pathset.rollback()
+                raise
+
+            if old_pathset is not None:
+                old_pathset.commit()
         finally:
             if cleanup_archive:
                 archive.unlink()
@@ -622,19 +639,43 @@ def _update(self, operation: Install | Update) -> int:
         return self._install(operation)
 
     def _remove(self, package: Package) -> int:
-        # If we have a VCS package, remove its source directory
+        result = self._uninstall(package)
+        if isinstance(result, int):
+            return result
+        result.commit()
+        return 0
+
+    def _uninstall(self, package: Package) -> int | UninstallPathSet:
+        """Stash an installed package's files.
+
+        Returns either an ``UninstallPathSet`` (builtin installer)
+        so the caller can ``.commit()`` to finalize the removal or
+        ``.rollback()`` to restore the files, or an ``int`` return code
+        (legacy pip path, or builtin path when nothing needs uninstalling).
+        ``-2`` means pip was interrupted; ``0`` means success.
+
+        The git VCS source directory (``<env>/src/<pkg>/``) is removed
+        eagerly and is not part of the returned pathset; it is not
+        restored on rollback.
+        """
         if package.source_type == "git":
             src_dir = self._env.path / "src" / package.name
             if src_dir.exists():
                 remove_directory(src_dir, force=True)
 
+        if self._use_builtin_uninstall:
+            pathset = uninstall_distribution(self._env, package.name)
+            # uninstall_distribution returns None when there is nothing to
+            # uninstall (not installed, outside env, in stdlib, missing
+            # RECORD). Treat that as a successful no-op.
+            return pathset if pathset is not None else 0
+
         try:
             return self.run_pip("uninstall", package.name, "-y")
         except EnvCommandError as e:
-            if "not installed" in str(e):
-                return 0
-
-            raise
+            if "not installed" not in str(e):
+                raise
+            return 0
 
     def _prepare_archive(
         self, operation: Install | Update, *, output_dir: Path | None = None