feat(pm): say when this process must restart to load the selected generation

A process that could not adopt a newly published dependency generation keeps
importing the old one. restart_needed() names that case (and stays silent for
dev venvs, Nix and anything not booted from a PM generation, so no false
restart prompts); adopt_selected() lets callers move onto the selection before
loading new code. The test helpers publish real generations and make the test
process run from one.

(cherry picked from commit 978abe8ec4b852778b8c63bcefdf141db51bc703)
(cherry picked from commit 28be27334adc531db59bf37a02a64acaaf47a2ee)

pm/environments_adopt.py

@@ -83,3 +83,35 @@ def adopt(previous: Path, selected: Path, running: Path) -> bool:
     os.environ["PATH"] = _replace(os.environ.get("PATH", ""), venv_bin_dir(previous), venv_bin_dir(selected))
     importlib.invalidate_caches()
     return True
+
+
+def _running_and_selected(project_root: Path) -> tuple[Path, Path] | None:
+    running = running_environment(project_root)
+    if running is None:
+        return None
+    try:
+        return running, selected_venv(project_root)
+    except (OSError, RuntimeError, ValueError):
+        return None
+
+
+def restart_needed(project_root: Path) -> str | None:
+    """Why this process must restart to load the selected generation, or None when it runs it.
+
+    Also None when the process does not run from one of this install's generations (a developer
+    venv, Nix): it never loads a PM selection, so a restart would not change what it imports.
+    The reason names the selected generation, so it changes with every new publication.
+    """
+    pair = _running_and_selected(project_root)
+    if pair is None or pair[0].resolve() == pair[1].resolve():
+        return None
+    running, selected = pair
+    return (f"dependency generation {selected.parent.name} was published after this process "
+            f"loaded {running.parent.name}")
+
+
+def adopt_selected(project_root: Path) -> bool:
+    """Move this process onto the selected generation. True when it already runs it, adopted it,
+    or runs from no generation of this install at all (there is nothing to adopt)."""
+    pair = _running_and_selected(project_root)
+    return pair is None or adopt(pair[0], pair[1], pair[0])

tests/hermes_cli/plugin_worker_support.py

@@ -114,6 +114,49 @@ class PluginWorld:
         assert result.returncode == 0, result.stderr
 
 
+def publish_plugins(world: PluginWorld, plugins: dict[str, list[str]]) -> Path:
+    """Enable exactly ``plugins`` (name → declared requirements) and publish the generation PM
+    builds for them. Each plugin imports its requirements, leaves an ``imported`` marker, and
+    ``request(specs)`` asks ``install_specs`` for more on its own behalf."""
+    from pm import client
+
+    for name, requirements in plugins.items():
+        plugin = world.home / "plugins" / name
+        if plugin.is_dir():
+            continue
+        plugin.mkdir(parents=True)
+        (plugin / "plugin.yaml").write_text(yaml.safe_dump(
+            {"name": name, "version": "1.0", "python_dependencies": requirements}), encoding="utf-8")
+        modules = [spec.split("=")[0].split("<")[0].split(">")[0].replace("-", "_") for spec in requirements]
+        (plugin / "__init__.py").write_text(
+            "".join(f"import {module}\n" for module in modules)
+            + "from pathlib import Path\nPath(__file__).with_name('imported').touch()\n"
+            "def register(ctx):\n    pass\n"
+            "def request(specs):\n    from tools.lazy_deps import install_specs\n    return install_specs(specs)\n",
+            encoding="utf-8")
+    (world.home / "config.yaml").write_text(yaml.safe_dump(
+        {"plugins": {"enabled": sorted(plugins), "disabled": []}}), encoding="utf-8")
+    client.sync_venv(explicit=True)
+    return world.selected()
+
+
+@pytest.fixture
+def boot(plugin_world, monkeypatch):
+    """``boot(environment)``: this test process imports from ``environment`` the way a Hermes process
+    booted on it does. Adoption rewrites sys.path and PATH (monkeypatch restores both); modules
+    imported from the world are forgotten afterwards, so a later test imports its own."""
+    from pm.environments import site_packages, venv_bin_dir
+
+    def run_from(environment: Path) -> None:
+        monkeypatch.setenv("PATH", os.pathsep.join([str(venv_bin_dir(environment)), os.defpath]))
+        monkeypatch.syspath_prepend(str(site_packages(environment)))
+
+    yield run_from
+    for name, module in list(sys.modules.items()):
+        if str(getattr(module, "__file__", None) or "").startswith(str(plugin_world.root)):
+            del sys.modules[name]
+
+
 @pytest.fixture
 def plugin_world(tmp_path, monkeypatch, isolated_python):
     from pm import client, paths

tests/pm/test_restart_needed.py

@@ -0,0 +1,23 @@
+"""restart_needed / adopt_selected against real generations: real uv, offline wheels, temp HERMES_HOME."""
+from __future__ import annotations
+
+from tests.hermes_cli.plugin_worker_support import (
+    boot as boot, isolated_python as isolated_python, plugin_world as plugin_world, publish_plugins)
+
+
+def test_a_process_on_a_superseded_generation_needs_a_restart_until_it_adopts(plugin_world, boot):
+    from pm.environments_adopt import adopt_selected, restart_needed
+
+    world = plugin_world
+    first = publish_plugins(world, {"base": ["plugin-proof-dep==1.0"]})
+    # This interpreter never booted from the install: nothing it could restart into.
+    assert restart_needed(world.core) is None
+    assert adopt_selected(world.core)
+    boot(first)
+    assert restart_needed(world.core) is None
+    second = publish_plugins(world, {"base": ["plugin-proof-dep==1.0"], "adds": ["plugin-proof-other"]})
+    assert second != first
+    reason = restart_needed(world.core)
+    assert reason and second.parent.name in reason
+    assert adopt_selected(world.core)
+    assert restart_needed(world.core) is None