Reuse installed packages for exact snapshot resets

conda_self/cli/main_reset.py

@@ -22,6 +22,7 @@ class Snapshot(Enum):
     """
 
     CURRENT = "current"
+    INSTALLER = "installer"  # Accepted only to explain how to migrate.
     INSTALLER_EXACT = "installer-exact"
     INSTALLER_UPDATED = "installer-updated"
     BASE_PROTECTION = "base-protection"
@@ -34,6 +35,8 @@ def display_name(self) -> str:
         match self:
             case Snapshot.CURRENT:
                 return "current"
+            case Snapshot.INSTALLER:
+                return "installer"
             case Snapshot.INSTALLER_EXACT:
                 return "installer-provided (exact)"
             case Snapshot.INSTALLER_UPDATED:
@@ -49,7 +52,7 @@ def file_path(self) -> Path | None:
                 return Path(sys.prefix, "conda-meta", RESET_FILE_INSTALLER)
             case Snapshot.BASE_PROTECTION:
                 return Path(sys.prefix, "conda-meta", RESET_FILE_BASE_PROTECTION)
-            case Snapshot.CURRENT:
+            case Snapshot.CURRENT | Snapshot.INSTALLER:
                 return None
 
 
@@ -73,6 +76,8 @@ def file_path(self) -> Path | None:
     currently installed versions (no downgrade).
     `base-protection` restores the `base` environment to the snapshot saved
     by `conda doctor --fix` before protecting base.
+    The old `installer` spelling is rejected with migration guidance. Choose
+    `installer-exact` or `installer-updated` explicitly.
 
     If not set, `conda self` will try to reset to the base-protection snapshot
     first, then to the installer-provided (preserving updates), and finally
@@ -112,12 +117,23 @@ def configure_parser(parser: argparse.ArgumentParser) -> None:
 
 def execute(args: argparse.Namespace) -> int:
     from conda.base.context import context
+    from conda.cli.common import stdout_json_success
+    from conda.exceptions import CondaValueError
     from conda.reporters import confirm_yn
 
     from ..query import permanent_dependencies
     from ..reset import names_from_explicit, reset
 
     snapshot: Snapshot | None = args.snapshot
+    if snapshot is Snapshot.INSTALLER:
+        raise CondaValueError(
+            "The '--snapshot installer' mode is no longer supported. "
+            "Use '--snapshot installer-exact' to restore the exact conda packages "
+            "recorded by the installer, or '--snapshot installer-updated' to keep "
+            "currently installed installer packages, conda plugins, and permanent "
+            "packages with their dependencies. 'installer-updated' does not update "
+            "packages or install missing packages. No reset was performed."
+        )
     reset_file: Path | None = None
 
     if snapshot is not None:
@@ -135,7 +151,7 @@ def execute(args: argparse.Namespace) -> int:
             f"Failed to reset to '{snapshot}'.\nRequired file {reset_file} not found."
         )
 
-    if not context.quiet:
+    if not context.json and not context.quiet:
         if snapshot is not None:
             print(WHAT_TO_EXPECT_SNAPSHOT.format(snapshot_name=snapshot.display_name))
         else:
@@ -146,7 +162,7 @@ def execute(args: argparse.Namespace) -> int:
         prompt += f" to the {snapshot.display_name} snapshot"
     confirm_yn(f"{prompt}?[y/n]:\n", default="no", dry_run=context.dry_run)
 
-    if not context.quiet:
+    if not context.json and not context.quiet:
         print("Resetting 'base' environment...")
 
     match snapshot:
@@ -160,7 +176,9 @@ def execute(args: argparse.Namespace) -> int:
         case _:
             reset(uninstallable_packages=permanent_dependencies(add_plugins=True))
 
-    if not context.quiet:
+    if context.json:
+        stdout_json_success()
+    elif not context.quiet:
         if snapshot is not None:
             print(SUCCESS_SNAPSHOT.format(snapshot_name=snapshot.display_name))
         else:

conda_self/reset.py

@@ -1,29 +1,166 @@
 from __future__ import annotations
 
+import re
 import sys
+from os.path import dirname, isfile
 from typing import TYPE_CHECKING
 
 from boltons.setutils import IndexedSet
+from conda import CondaError, CondaExitZero, CondaMultiError
 from conda.base.constants import EXPLICIT_MARKER
 from conda.base.context import context
+from conda.common.io import dashlist
+from conda.common.path import get_major_minor_version
 from conda.core.link import PrefixSetup, UnlinkLinkTransaction
+from conda.core.package_cache_data import PackageCacheData
 from conda.core.prefix_data import PrefixData
 from conda.core.solve import diff_for_unlink_link_precs
-from conda.gateways.disk.read import yield_lines
-from conda.misc import get_package_records_from_explicit
+from conda.exceptions import ChecksumMismatchError, CondaSignalInterrupt, ParseError
+from conda.gateways.disk.read import compute_sum, yield_lines
+from conda.misc import _match_specs_from_explicit, get_package_records_from_explicit
+from conda.models.enums import NoarchType
 from conda.models.match_spec import MatchSpec
+from conda_package_handling.exceptions import InvalidArchiveError
 
 if TYPE_CHECKING:
     from pathlib import Path
 
+    from conda.models.records import PackageRecord
+
+
+def records_from_snapshot(
+    prefix: str, snapshot_content: list[str]
+) -> tuple[IndexedSet, tuple[MatchSpec, ...]]:
+    """Return package records for the snapshot and MatchSpecs to install or relink."""
+    entries = tuple(line for line in snapshot_content if line != EXPLICIT_MARKER)
+    try:
+        specs = tuple(_match_specs_from_explicit(entries))
+    except (ParseError, ValueError, IndexError, re.error):
+        raise ParseError("Could not parse a package URL in the snapshot.") from None
+    prefix_data = PrefixData(prefix)
+    installed_python = prefix_data.get("python", None)
+    snapshot_python = next((spec for spec in specs if spec.name == "python"), None)
+    snapshot_python_version = (
+        snapshot_python.get_exact_value("version")
+        if snapshot_python is not None
+        else None
+    )
+    relink_noarch_python = bool(
+        installed_python is not None
+        and snapshot_python_version
+        and get_major_minor_version(installed_python.version)
+        != get_major_minor_version(snapshot_python_version)
+    )
+
+    records: list[PackageRecord | None] = [None] * len(entries)
+    unresolved_entries: list[str] = []
+    unresolved_specs: list[MatchSpec] = []
+    unresolved_indices: list[int] = []
+
+    for index, (entry, spec) in enumerate(zip(entries, specs, strict=True)):
+        installed_record = next(iter(prefix_data.query(spec)), None)
+        needs_python_relink = bool(
+            installed_record is not None
+            and relink_noarch_python
+            and installed_record.noarch == NoarchType.python
+        )
+        if installed_record is not None and not needs_python_relink:
+            records[index] = installed_record
+            continue
+
+        unresolved_entries.append(entry)
+        unresolved_specs.append(spec)
+        unresolved_indices.append(index)
+
+    if unresolved_entries:
+        try:
+            fetched_records = tuple(
+                get_package_records_from_explicit(unresolved_entries)
+            )
+            checksum_errors = []
+            for spec, record in zip(unresolved_specs, fetched_records, strict=True):
+                url = spec.get_exact_value("url")
+                if not url or not url.startswith("file:"):
+                    continue
+                archive = record.package_tarball_full_path
+                if not isfile(archive):
+                    continue
+                # Conda trusts the requested checksum when copying a local package.
+                for checksum_type in ("md5", "sha256"):
+                    expected_checksum = spec.get_exact_value(checksum_type)
+                    if expected_checksum is None:
+                        continue
+                    actual_checksum = compute_sum(archive, checksum_type)
+                    if actual_checksum != expected_checksum:
+                        # Keep the extracted cache usable only for its actual bytes.
+                        setattr(record, checksum_type, actual_checksum)
+                        PackageCacheData(dirname(archive)).insert(record)
+                        checksum_errors.append(
+                            ChecksumMismatchError(
+                                url,
+                                archive,
+                                checksum_type,
+                                expected_checksum,
+                                actual_checksum,
+                            )
+                        )
+            if checksum_errors:
+                raise CondaMultiError(checksum_errors)
+        except (CondaError, InvalidArchiveError) as error:
+            pending_errors = [error]
+            found_error = False
+            while pending_errors:
+                nested_error = pending_errors.pop()
+                if isinstance(nested_error, CondaMultiError):
+                    pending_errors.extend(nested_error.errors)
+                elif type(nested_error) is RuntimeError and str(
+                    nested_error
+                ).startswith(
+                    f"{InvalidArchiveError.__module__}."
+                    f"{InvalidArchiveError.__qualname__}: "
+                ):
+                    # Conda 26.7 extraction workers wrap unpicklable errors in
+                    # RuntimeError, retaining only the qualified type and message.
+                    found_error = True
+                elif not isinstance(
+                    nested_error, (CondaError, InvalidArchiveError)
+                ) or isinstance(nested_error, (CondaExitZero, CondaSignalInterrupt)):
+                    raise
+                else:
+                    found_error = True
+            if not found_error:
+                raise
+
+            packages = dashlist(
+                spec.get_exact_value("fn") or spec.name for spec in unresolved_specs
+            )
+            raise CondaError(
+                "Could not make all conda packages required by the selected "
+                "snapshot available in a package cache.\n"
+                "Required conda packages:%(packages)s\n"
+                "The target environment was not changed. Some conda packages "
+                "may have been downloaded and extracted into a package cache. "
+                "Ensure each listed package is available in a package cache or "
+                "can be downloaded from the URL in the snapshot, verified using "
+                "its recorded checksum when present, and extracted, then retry.",
+                packages=packages,
+            ) from None
+
+        for index, record in zip(unresolved_indices, fetched_records, strict=True):
+            records[index] = record
+
+    return IndexedSet(record for record in records if record is not None), tuple(
+        unresolved_specs
+    )
+
 
 def names_from_explicit(path: Path) -> set[str]:
-    """Extract package names from a CEP-23 ``@EXPLICIT`` file without fetching.
+    """Extract package names from a CEP 23 explicit spec file without downloading.
 
     Parses each URL line with :class:`~conda.models.match_spec.MatchSpec`,
-    which reads ``name``/``version``/``build`` from the tarball filename and
-    strips any ``#md5=…``/``#sha256=…`` checksum fragment as a comment.  No
-    network access, unlike :func:`conda.misc.get_package_records_from_explicit`.
+    which reads ``name``/``version``/``build`` from the package filename and
+    strips any checksum fragment as a comment. No network access, unlike
+    :func:`conda.misc.get_package_records_from_explicit`.
     """
     return {
         MatchSpec(line).name for line in yield_lines(path) if line != EXPLICIT_MARKER
@@ -37,17 +174,22 @@ def reset(
 ):
     if snapshot:
         snapshot_content = list(yield_lines(snapshot))
-        packages_in_reset_env = IndexedSet(
-            get_package_records_from_explicit(snapshot_content)
+        packages_in_reset_env, unresolved_specs = records_from_snapshot(
+            prefix, snapshot_content
         )
         packages_to_remove, packages_to_install = diff_for_unlink_link_precs(
-            prefix, packages_in_reset_env
+            prefix,
+            packages_in_reset_env,
+            specs_to_add=unresolved_specs,
+            force_reinstall=True,
         )
         if not packages_to_remove and not packages_to_install:
-            print(
-                "Nothing to do. "
-                "Packages in target environment match the selected snapshot."
-            )
+            if not context.json and not context.quiet:
+                print(
+                    "Nothing to do. "
+                    "The conda packages in the target environment match the "
+                    "selected snapshot."
+                )
             return
     else:
         installed = sorted(PrefixData(prefix).iter_records(), key=lambda x: x.name)

docs/configuration.md

@@ -30,7 +30,7 @@ use conda's `@EXPLICIT` format (a list of exact package URLs).
 | File | Created by | Purpose |
 |------|-----------|---------|
 | `base-protection-state.explicit.txt` | `conda doctor base-protection --fix` | Pre-protection state of base |
-| `installer-state.explicit.txt` | Installer (e.g. Miniforge) | Original installer state |
+| `initial-state.explicit.txt` | Installer (e.g. Miniforge) | Original installer state |
 
 These files are used by `conda self reset --snapshot <type>` to
 restore base to a known state without running the solver.
@@ -41,7 +41,7 @@ restore base to a known state without running the solver.
 |----------|-------|-------------|
 | `DEFAULT_ENV_NAME` | `"default"` | Name of the environment created when cloning base |
 | `SNAPSHOT_FILE_BASE_PROTECTION` | `"base-protection-state.explicit.txt"` | Snapshot filename for base protection |
-| `RESET_FILE_INSTALLER` | `"installer-state.explicit.txt"` | Snapshot filename from installer |
+| `RESET_FILE_INSTALLER` | `"initial-state.explicit.txt"` | Snapshot filename from installer |
 | `SELF_PERMANENT_PACKAGES_SETTING` | `"self_permanent_packages"` | Name of the condarc setting |
 
 ## Environment variables

docs/features.md

@@ -93,15 +93,22 @@ pre-protection state. This snapshot can be used to restore base:
 
 ```bash
 conda self reset                         # auto-detect best snapshot
-conda self reset --snapshot installer    # reset to installer state
+conda self reset --snapshot installer-exact    # reset to installer state
 conda self reset --snapshot base-protection  # reset to protection snapshot
 conda self reset --snapshot current      # strip to essentials only
 ```
 
+The old `--snapshot installer` option reports a migration error. Use
+`installer-exact` to restore the exact snapshot, or `installer-updated` to
+retain installed versions of installer packages alongside conda, conda-self,
+installed conda plugins, configured permanent packages, and their dependencies.
+See {doc}`guides/resetting-base` for migration guidance and exact-reset
+requirements.
+
 Snapshots are stored as `@EXPLICIT` files in `conda-meta/`:
 
 - `base-protection-state.explicit.txt` -- saved by `conda doctor --fix`
-- `installer-state.explicit.txt` -- saved by the installer (if available)
+- `initial-state.explicit.txt` -- saved by the installer (if available)
 
 ## Health check integration