exp run: error out when trying to resume from intermediate checkpoint w/no modifications (#5829)

* apply: allow applying exp with multiple possible branches

* default to new checkpoint branch after applying ambiguous commit

* output shas in addition to exp names by default

* update show tests

* git: allow force checkout in detach_head context

* exp run: error when trying to resume from a non-head checkpoint without
modifications

* add test for apply/resume from intermediate checkpoint

* list some possible heads when multiple matches

dvc/command/experiments.py

@@ -172,7 +172,10 @@ def _collect_rows(
                 else:
                     tree = "└──"
                 new_checkpoint = True
-            name = exp.get("name", rev[:7])
+            if "name" in exp:
+                name = f"{rev[:7]} [[bold]{exp['name']}[/]]"
+            else:
+                name = rev[:7]
 
         row = [
             name,

dvc/repo/experiments/__init__.py

@@ -194,7 +194,7 @@ class Experiments:
                         if branch:
                             branch_name = ExpRefInfo.from_ref(branch).name
                         else:
-                            branch_name = ""
+                            branch_name = f"{resume_rev[:7]}"
                         if self.scm.is_dirty():
                             logger.info(
                                 "Modified checkpoint experiment based on "
@@ -202,6 +202,39 @@ class Experiments:
                                 branch_name,
                             )
                             branch = None
+                        elif (
+                            not branch
+                            or self.scm.get_ref(branch) != resume_rev
+                        ):
+                            msg = [
+                                (
+                                    "Nothing to do for unchanged checkpoint "
+                                    f"'{resume_rev[:7]}'. "
+                                )
+                            ]
+                            if branch:
+                                msg.append(
+                                    "To resume from the head of this "
+                                    "experiment, use "
+                                    f"'dvc exp apply {branch_name}'."
+                                )
+                            else:
+                                names = [
+                                    ref_info.name
+                                    for ref_info in exp_refs_by_rev(
+                                        self.scm, resume_rev
+                                    )
+                                ]
+                                if len(names) > 3:
+                                    names[3:] = [
+                                        f"... ({len(names) - 3} more)"
+                                    ]
+                                msg.append(
+                                    "To resume an experiment containing this "
+                                    "checkpoint, apply one of these heads:\n"
+                                    "\t{}".format(", ".join(names))
+                                )
+                            raise DvcException("".join(msg))
                         else:
                             logger.info(
                                 "Existing checkpoint experiment '%s' will be "
@@ -237,11 +270,15 @@ class Experiments:
                         baseline_rev[:7],
                     )
                 finally:
-                    # Reset any of our changes before prior unstashing
                     if resume_rev:
+                        # NOTE: this set_ref + reset() is equivalent to
+                        # `git reset orig_head` (our SCM reset() only operates
+                        # on HEAD rather than any arbitrary commit)
                         self.scm.set_ref(
                             "HEAD", orig_head, message="dvc: restore HEAD"
                         )
+                        self.scm.reset()
+                    # Revert any of our changes before prior unstashing
                     self.scm.reset(hard=True)
 
         return stash_rev
@@ -447,22 +484,34 @@ class Experiments:
         """
         assert resume_rev
 
-        allow_multiple = "params" in kwargs
-        branch: Optional[str] = self.get_branch_by_rev(
-            resume_rev, allow_multiple=allow_multiple
-        )
-        if not branch:
-            raise DvcException(
-                "Could not find checkpoint experiment " f"'{resume_rev[:7]}'"
+        branch: Optional[str] = None
+        try:
+            allow_multiple = bool(kwargs.get("params", None))
+            branch = self.get_branch_by_rev(
+                resume_rev, allow_multiple=allow_multiple
             )
+            if not branch:
+                raise DvcException(
+                    "Could not find checkpoint experiment "
+                    f"'{resume_rev[:7]}'"
+                )
+            baseline_rev = self._get_baseline(branch)
+        except MultipleBranchError as exc:
+            baselines = {
+                info.baseline_sha
+                for info in exc.ref_infos
+                if info.baseline_sha
+            }
+            if len(baselines) == 1:
+                baseline_rev = baselines.pop()
+            else:
+                raise
 
-        baseline_rev = self._get_baseline(branch)
         logger.debug(
-            "Resume from checkpoint '%s' with baseline '%s'",
-            resume_rev,
+            "Checkpoint run from '%s' with baseline '%s'",
+            resume_rev[:7],
             baseline_rev,
         )
-
         return self._stash_exp(
             *args,
             resume_rev=resume_rev,
@@ -707,10 +756,10 @@ class Experiments:
 
         # NOTE: the stash commit to be popped already contains all the current
         # workspace changes plus CLI modifed --params changes.
-        # `reset --hard` here will not lose any data (pop without reset would
-        # result in conflict between workspace params and stashed CLI params).
-        self.scm.reset(hard=True)
-        with self.scm.detach_head(entry.rev):
+        # `checkout --force` here will not lose any data (popping stash commit
+        # will result in conflict between workspace params and stashed CLI
+        # params, but we always want the stashed version).
+        with self.scm.detach_head(entry.rev, force=True):
             rev = self.stash.pop()
             self.scm.set_ref(EXEC_BASELINE, entry.baseline_rev)
             if entry.branch:
@@ -797,7 +846,10 @@ class Experiments:
         if not ref_infos:
             return None
         if len(ref_infos) > 1 and not allow_multiple:
-            raise MultipleBranchError(rev)
+            for ref_info in ref_infos:
+                if self.scm.get_ref(str(ref_info)) == rev:
+                    return str(ref_info)
+            raise MultipleBranchError(rev, ref_infos)
         return str(ref_infos[0])
 
     def get_exact_name(self, rev: str):

dvc/repo/experiments/apply.py

@@ -32,7 +32,9 @@ def apply(repo, rev, force=True, **kwargs):
         raise InvalidExpRevError(rev) from exc
 
     stash_rev = exp_rev in exps.stash_revs
-    if not stash_rev and not exps.get_branch_by_rev(exp_rev):
+    if not stash_rev and not exps.get_branch_by_rev(
+        exp_rev, allow_multiple=True
+    ):
         raise InvalidExpRevError(exp_rev)
 
     # Note that we don't use stash_workspace() here since we need finer control

dvc/repo/experiments/base.py

@@ -74,12 +74,13 @@ class InvalidExpRevError(InvalidArgumentError):
 
 
 class MultipleBranchError(DvcException):
-    def __init__(self, rev):
+    def __init__(self, rev, ref_infos):
         super().__init__(
             f"Ambiguous commit '{rev[:7]}' belongs to multiple experiment "
             "branches."
         )
         self.rev = rev
+        self.ref_infos = ref_infos
 
 
 class ApplyConflictError(InvalidArgumentError):

dvc/scm/git/__init__.py

@@ -426,7 +426,7 @@ class Git(Base):
             commit = self.resolve_commit(parent)
 
     @contextmanager
-    def detach_head(self, rev: Optional[str] = None):
+    def detach_head(self, rev: Optional[str] = None, force: bool = False):
         """Context manager for performing detached HEAD SCM operations.
 
         Detaches and restores HEAD similar to interactive git rebase.
@@ -440,7 +440,7 @@ class Git(Base):
             rev = "HEAD"
         orig_head = self.get_ref("HEAD", follow=False)
         logger.debug("Detaching HEAD at '%s'", rev)
-        self.checkout(rev, detach=True)
+        self.checkout(rev, detach=True, force=force)
         try:
             yield self.get_ref("HEAD")
         finally:

tests/func/experiments/test_checkpoints.py

@@ -160,3 +160,27 @@ def test_resume_branch(tmp_dir, scm, dvc, checkpoint_stage, workspace):
     assert branch_rev == dvc.experiments.scm.gitpython.repo.git.merge_base(
         checkpoint_a, checkpoint_b
     )
+
+
+@pytest.mark.parametrize("workspace", [True, False])
+def test_resume_non_head_checkpoint(
+    tmp_dir, scm, dvc, checkpoint_stage, workspace
+):
+    orig_head = scm.get_rev()
+    results = dvc.experiments.run(
+        checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace
+    )
+    checkpoint_head = first(results)
+    orig_branch = dvc.experiments.get_branch_by_rev(checkpoint_head)
+
+    rev = list(scm.branch_revs(checkpoint_head, orig_head))[-1]
+    dvc.experiments.apply(rev)
+
+    with pytest.raises(DvcException):
+        dvc.experiments.run(checkpoint_stage.addressing, tmp_dir=not workspace)
+
+    results = dvc.experiments.run(
+        checkpoint_stage.addressing, params=["foo=100"], tmp_dir=not workspace
+    )
+    new_head = first(results)
+    assert orig_branch != dvc.experiments.get_branch_by_rev(new_head)

tests/func/experiments/test_show.py

@@ -111,6 +111,7 @@ def test_show_checkpoint(
     for i, rev in enumerate(checkpoints):
         if i == 0:
             name = dvc.experiments.get_exact_name(rev)
+            name = f"{rev[:7]} [{name}]"
             fs = "╓"
         elif i == len(checkpoints) - 1:
             name = rev[:7]
@@ -155,7 +156,7 @@ def test_show_checkpoint_branch(
     for rev in (checkpoint_a, checkpoint_b):
         ref = dvc.experiments.get_branch_by_rev(rev)
         ref_info = ExpRefInfo.from_ref(ref)
-        name = ref_info.name
+        name = f"{rev[:7]} [{ref_info.name}]"
         assert f"╓ {name}" in cap.out
     assert f"({branch_rev[:7]})" in cap.out