fix(pm): rebuild side environments when the selected interpreter moves

A venv records its base interpreter path in pyvenv.cfg, not just a version.
The generation identity hashed only the pinned Python artifact, so the same
pin realized in a different tools store (a fresh HERMES_RUNTIME_DIR, a CI
setup-pm store) reused a venv whose interpreter no longer existed or belonged
to another store. Key the identity on the resolved selected interpreter and
build against it explicitly; a failed replacement still keeps the prior
generation selected.
This commit is contained in:
ethernet
2026-09-24 10:28:25 -04:00
parent 8bef97fa16
commit 7b68fa7700
2 changed files with 89 additions and 14 deletions

View File

@@ -196,7 +196,7 @@ def ensure_environment(
if executable is not None:
_tool(Path("unused/python"), executable) # Validate before any write.
def build(generation: Path) -> Path:
def build(generation: Path, base_python: Path) -> Path:
(generation / "pyproject.toml").write_text(
'[project]\nname = "hermes-side-environment"\nversion = "0"\n'
'requires-python = ">=3.11"\ndependencies = '
@@ -207,8 +207,8 @@ def ensure_environment(
seed = root / previous["generation"] / "uv.lock"
if seed.is_file():
shutil.copyfile(seed, generation / "uv.lock")
return build_environment(source=generation, out=generation / "venv", frozen=False,
explicit=explicit, timeout=timeout)
return build_environment(source=generation, out=generation / "venv", python=base_python,
frozen=False, explicit=explicit, timeout=timeout)
return _ensure_generation(name, root, {"requirements": requirements}, build,
record={"requirements": requirements}, explicit=explicit,
@@ -236,16 +236,17 @@ def ensure_project_environment(
except FileNotFoundError as exc:
raise InstallError("venv", f"locked project environment needs {project / manifest}") from exc
def build(generation: Path) -> Path:
return build_environment(source=project, out=generation / "venv", extras=extras, groups=groups,
no_install_project=True, frozen=True, explicit=explicit, timeout=timeout)
def build(generation: Path, base_python: Path) -> Path:
return build_environment(source=project, out=generation / "venv", python=base_python,
extras=extras, groups=groups, no_install_project=True,
frozen=True, explicit=explicit, timeout=timeout)
return _ensure_generation(name, root, {"manifests": manifests, "extras": extras, "groups": groups},
build, record={"extras": extras, "groups": groups}, explicit=explicit)
def _ensure_generation(
name: str, root: Path, inputs: dict, build: Callable[[Path], Path], *, record: dict, explicit: bool,
name: str, root: Path, inputs: dict, build: Callable[[Path, Path], Path], *, record: dict, explicit: bool,
executable: str | None = None,
) -> Path:
"""Select the generation whose inputs match, building one only when none does.
@@ -254,6 +255,7 @@ def _ensure_generation(
The prior generation survives both successful replacement and failed builds.
"""
from hermes_cli.runtime_state import _lock
from pm._uv import _toolchain
from pm.install import _refuse_lazy, lazy_installs_allowed
from pm.lock import Lockfile, _write
from pm import paths
@@ -263,17 +265,28 @@ def _ensure_generation(
target = current_target()
inputs = {**inputs, "python": lock.version("python"), "target": target,
"artifacts": [item["sha256"] for item in lock.artifacts("python", target)]}
identity = hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest()
def current() -> Path | None:
def selected_python() -> Path | None:
tools = _toolchain(realize=False)
return tools[1].resolve() if tools is not None else None
def identity(base_python: Path) -> str:
# The same pinned artifact can live in different stores. A venv's
# pyvenv.cfg keeps the original interpreter path, not just its version.
return hashlib.sha256(json.dumps({**inputs, "interpreter": str(base_python)},
sort_keys=True).encode()).hexdigest()
def current(base_python: Path | None) -> Path | None:
if base_python is None:
return None
selected = _selection(root)
python = environment_python(name, root=root)
if (selected.get("inputs") == identity and python is not None
if (selected.get("inputs") == identity(base_python) and python is not None
and (executable is None or _tool(python, executable) is not None)):
return python
return None
existing = current()
existing = current(selected_python())
if existing is not None:
return existing
if not explicit and not lazy_installs_allowed():
@@ -281,16 +294,23 @@ def _ensure_generation(
root.mkdir(parents=True, exist_ok=True)
with (root / ".install.lock").open("a+b") as mutex:
_lock(mutex.fileno(), wait=True)
existing = current()
base_python = selected_python()
existing = current(base_python)
if existing is not None:
return existing
if base_python is None:
tools = _toolchain(explicit=explicit)
if tools is None:
raise InstallError(name, "PM's pinned toolchain is unavailable")
base_python = tools[1].resolve()
generation = root / f"gen-{uuid.uuid4().hex}"
generation.mkdir()
try:
python = build(generation)
python = build(generation, base_python)
if executable is not None and _tool(python, executable) is None:
raise InstallError(name, f"installed requirements do not provide {executable!r}")
_write(root / "active.json", {"generation": generation.name, "inputs": identity, **record})
_write(root / "active.json", {"generation": generation.name,
"inputs": identity(base_python), **record})
except BaseException:
shutil.rmtree(generation, ignore_errors=True)
raise

View File

@@ -103,6 +103,61 @@ def test_all_uv_commands_keep_the_pm_interpreter(installed_uv, monkeypatch):
assert dict(os.environ) == before
def test_project_environment_replaces_generation_when_pinned_python_moves(installed_uv, tmp_path, monkeypatch):
"""An unchanged dependency pin cannot reuse a venv made by another tools store."""
from pm import operations
import pm.registry as registry
from tests.pm._fixtures import _run, _wheel, stage_host_python
root, uv, facts, target, digest = installed_uv
store = root / "store"
from pm.store import tree_digest
facts.record("uv", "test", uv.parent.name, {}, store, target=target,
artifacts=[digest], digest=tree_digest(uv.parent))
class FixturePython(Python):
binary_rel = {"win32": "Scripts/python.exe", "posix": "bin/python"}
monkeypatch.setitem(registry._packages, "python", FixturePython())
interpreters = [stage_host_python(store / name / "bin" / "python")
for name in ("python-a", "python-b")]
project = tmp_path / "project"
project.mkdir()
wheel = _wheel(tmp_path, "side_dep")
(project / "pyproject.toml").write_text(
'[project]\nname="pm-generation-proof"\nversion="1"\nrequires-python=">=3.11"\n'
'dependencies=["side-dep==1.0"]\n[tool.uv]\npackage=false\nno-index=true\n'
f'find-links=["{wheel.parent.as_posix()}"]\n', encoding="utf-8",
)
env = {key: value for key, value in os.environ.items()
if not key.startswith(("UV_", "PYTHON")) and key != "VIRTUAL_ENV"}
env.update(UV_CACHE_DIR=str(tmp_path / "cache"), UV_PYTHON_DOWNLOADS="never")
_run([str(uv), "lock", "--python", str(interpreters[0])], cwd=project, env=env)
locked = (project / "uv.lock").read_bytes()
selection_root = tmp_path / "selection"
chosen = []
for interpreter in (*interpreters, interpreters[0]):
facts.record("python", "test", interpreter.parent.parent.name, {}, store,
target=target, artifacts=[digest], digest=tree_digest(interpreter.parent.parent))
selected = operations.ensure_project_environment(
"test-environment", project, root=selection_root, explicit=True)
chosen.append(selected)
assert Path(_run([str(selected), "-I", "-c",
"import sys, side_dep; print(sys.base_prefix)"], cwd=project, env=env)) == interpreter.parent.parent
assert (project / "uv.lock").read_bytes() == locked
assert chosen[0] != chosen[1] != chosen[2]
# A failed replacement must leave the previously selected environment intact.
facts.record("python", "test", interpreters[1].parent.parent.name, {}, store,
target=target, artifacts=[digest], digest=tree_digest(interpreters[1].parent.parent))
active = (selection_root / "active.json").read_bytes()
monkeypatch.setattr(operations, "build_environment", lambda **kwargs: (_ for _ in ()).throw(RuntimeError("build failed")))
with pytest.raises(RuntimeError, match="build failed"):
operations.ensure_project_environment("test-environment", project, root=selection_root, explicit=True)
assert (selection_root / "active.json").read_bytes() == active
assert operations.environment_python("test-environment", root=selection_root) == chosen[-1]
def test_uv_refuses_discovery_when_pm_python_is_missing(installed_uv, monkeypatch):
root, _, facts, target, digest = installed_uv
ensure = importlib.import_module("pm.install")