fix(update): repoint Windows managed runtime without renaming live venv

Cherry-pick of helix4u's #88836 (75742db5a42) resolved onto current main.

The POSIX runtime cutover parks the live venv with a directory rename. On
Windows that can never succeed from `hermes update`: the updater executes
from venv\Scripts\python.exe and Windows keeps that image mapped, so the
rename fails with WinError 5 on every attempt (#93032). Instead, atomically
replace the live venv's pyvenv.cfg with the candidate's: Scripts\python.exe
is a launcher that reads `home` on every start, so every fresh process runs
the new generation while nothing touches the locked directory. A failed
post-cutover smoke restores the original config; a failed rollback keeps
the generation the live config now references.
This commit is contained in:
Gille
2026-09-17 00:18:14 -07:00
committed by Teknium
parent 032cf8438f
commit 5a49d1de3d
2 changed files with 274 additions and 7 deletions

View File

@@ -2,7 +2,11 @@
The Python backing the install is shared by every Hermes profile because the checkout's ``venv``
is shared. Runtime repair therefore uses an install-scoped store under
``<checkout>/.hermes-runtime/python``. A vulnerable interpreter is never reinstalled in place.
``<checkout>/.hermes-runtime/python``. A vulnerable interpreter is never reinstalled in place: a
new immutable Python generation is provisioned and a relocatable sibling venv built and smoke-tested
from it. POSIX installs cut over with same-filesystem directory renames; Windows installs
atomically repoint the live venv's ``pyvenv.cfg`` at the new generation instead, because the
updater itself executes from that venv and Windows refuses to rename it.
"""
from __future__ import annotations
@@ -754,6 +758,72 @@ def _cut_over_candidate(
raise
def _replace_file_atomically(path: Path, data: bytes) -> None:
"""Replace *path* from a same-directory temporary file."""
token = f"{os.getpid()}-{uuid.uuid4().hex[:8]}"
temporary = path.with_name(f".{path.name}.runtime-{token}.tmp")
try:
with temporary.open("xb") as handle:
handle.write(data)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
finally:
try:
temporary.unlink()
except FileNotFoundError:
pass
def _cut_over_windows_runtime_config(
candidate: Path,
*,
live: Path,
) -> tuple[bool, bool, SQLiteRuntimeInfo | None, str]:
"""Repoint a live Windows venv without renaming its locked directory.
``venv\\Scripts\\python.exe`` is a launcher that reads ``home`` from ``pyvenv.cfg`` on every
start, so replacing that text file redirects every fresh process to the candidate generation
while the updater keeps executing from the directory Windows refuses to rename (#93032).
The second return value reports whether the live config still references the candidate
generation. Callers must preserve that generation if a failed smoke test could not restore
the original config.
"""
live_config = live / "pyvenv.cfg"
candidate_config = candidate / "pyvenv.cfg"
try:
original = live_config.read_bytes()
replacement = candidate_config.read_bytes()
except OSError as exc:
return False, False, None, f"could not read venv runtime config: {exc}"
try:
_replace_file_atomically(live_config, replacement)
except OSError as exc:
return False, False, None, f"could not repoint the existing venv: {exc}"
try:
healthy, detail, info = _smoke_candidate_venv(live)
except Exception as exc:
healthy, detail, info = False, f"candidate smoke raised: {exc}", None
if healthy:
return True, True, info, ""
try:
_replace_file_atomically(live_config, original)
except OSError as rollback_error:
return (
False,
True,
info,
"post-cutover smoke failed "
f"({detail}); runtime-config rollback failed ({rollback_error})",
)
return False, False, info, f"post-cutover smoke failed: {detail}"
def _acquire_repair_lock(runtime_root: Path) -> _RepairLock | None:
"""Acquire an OS-held install lock that is released on process exit."""
runtime_root.mkdir(parents=True, exist_ok=True)
@@ -793,6 +863,10 @@ def _release_repair_lock(lock: _RepairLock) -> None:
os.close(lock.fd)
def _windows_runtime_holders() -> tuple[bool, str]:
if platform.system() != "Windows":
return False, ""
@@ -941,6 +1015,8 @@ def _result(
return RuntimeRepairResult(status, detail, sqlite_before=current.sqlite_version_string, **extra)
def _repair_windows_preflight(
root: Path, live: Path, current: SQLiteRuntimeInfo) -> RuntimeRepairResult | None:
"""Defer the repair when Windows holders make the venv rename impossible; else ``None``."""
@@ -1004,12 +1080,19 @@ def _repair_under_lock(
"failed", current, str(exc),
sqlite_after=candidate_info.sqlite_version_string)
cut_over, backup, final_info, cutover_detail = _cut_over_candidate(
candidate, project_root=root, live=live)
backup = None
generation_in_use = False
if platform.system() == "Windows":
cut_over, generation_in_use, final_info, cutover_detail = _cut_over_windows_runtime_config(
candidate, live=live)
else:
cut_over, backup, final_info, cutover_detail = _cut_over_candidate(
candidate, project_root=root, live=live)
if not cut_over:
if backup is None:
_remove_tree(candidate, boundary=runtime_root)
_remove_tree(generation, boundary=managed_python_install_dir(root))
if not generation_in_use:
_remove_tree(generation, boundary=managed_python_install_dir(root))
return _result(
"failed", current, cutover_detail,
sqlite_after=final_info.sqlite_version_string if final_info is not None else "",
@@ -1020,16 +1103,21 @@ def _repair_under_lock(
f"(SQLite {current.sqlite_version_string} → {final_version})")
if backup is not None and backup.exists():
_remove_tree(backup, boundary=root)
elif backup is None:
# Windows: the live venv now points at the generation; the staging venv is spent.
_remove_tree(candidate, boundary=runtime_root)
return _result("repaired", current, sqlite_after=final_version, backup_venv=backup)
def repair_vulnerable_runtime(
uv_bin: str, *, project_root: Path | None = None, venv_dir: Path | None = None
) -> RuntimeRepairResult:
"""Replace a vulnerable install venv without mutating it in place.
"""Replace a vulnerable install venv without mutating its packages in place.
Every failure before cutover leaves the live venv untouched. Rename or post-cutover smoke
failures restore the parked venv synchronously.
Every failure before cutover leaves the live venv untouched. POSIX cuts over with directory
renames and restores the parked venv synchronously on failure; Windows repoints the live
venv's ``pyvenv.cfg`` instead (the updater runs from that venv, so the directory can never
be renamed) and restores the original config on a failed smoke.
"""
root = Path(project_root) if project_root is not None else _PROJECT_ROOT
live = Path(venv_dir) if venv_dir is not None else _default_live_venv(root)

View File

@@ -0,0 +1,179 @@
"""Native-Windows coverage for managed runtime cutover."""
from __future__ import annotations
import subprocess
import venv
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import pytest
def _python(venv_dir: Path) -> Path:
return venv_dir / "Scripts" / "python.exe"
@pytest.mark.windows_only
def test_runtime_config_cutover_does_not_rename_running_venv(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from hermes_cli import managed_uv
live = tmp_path / "venv"
candidate = tmp_path / "candidate"
venv.EnvBuilder(with_pip=False).create(live)
venv.EnvBuilder(with_pip=False).create(candidate)
candidate_config = candidate / "pyvenv.cfg"
replacement = candidate_config.read_bytes() + b"runtime-cutover = candidate\n"
candidate_config.write_bytes(replacement)
info = SimpleNamespace(sqlite_version_string="3.53.1")
monkeypatch.setattr(
managed_uv,
"_smoke_candidate_venv",
lambda target: (True, "", info),
)
process = subprocess.Popen(
[str(_python(live)), "-c", "import time; time.sleep(30)"],
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
try:
ok, runtime_in_use, final_info, detail = (
managed_uv._cut_over_windows_runtime_config(candidate, live=live)
)
assert process.poll() is None
assert ok is True
assert runtime_in_use is True
assert final_info is info
assert detail == ""
assert (live / "pyvenv.cfg").read_bytes() == replacement
assert live.is_dir()
probe = subprocess.run(
[str(_python(live)), "-I", "-c", "print('repointed')"],
capture_output=True,
text=True,
check=False,
timeout=10,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
assert probe.returncode == 0
assert probe.stdout.strip() == "repointed"
finally:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=5)
@pytest.mark.windows_only
def test_runtime_config_cutover_rolls_back_failed_live_smoke(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from hermes_cli import managed_uv
live = tmp_path / "venv"
candidate = tmp_path / "candidate"
live.mkdir()
candidate.mkdir()
original = b"home = old-runtime\n"
(live / "pyvenv.cfg").write_bytes(original)
(candidate / "pyvenv.cfg").write_bytes(b"home = new-runtime\n")
monkeypatch.setattr(
managed_uv,
"_smoke_candidate_venv",
lambda target: (False, "core import smoke failed", None),
)
ok, runtime_in_use, info, detail = managed_uv._cut_over_windows_runtime_config(
candidate, live=live
)
assert ok is False
assert runtime_in_use is False
assert info is None
assert detail == "post-cutover smoke failed: core import smoke failed"
assert (live / "pyvenv.cfg").read_bytes() == original
@pytest.mark.windows_only
def test_runtime_repair_uses_config_cutover_on_windows(tmp_path: Path) -> None:
from hermes_cli.managed_uv import repair_vulnerable_runtime
from hermes_cli.sqlite_runtime import SQLiteRuntimeInfo
root = tmp_path / "checkout"
root.mkdir()
(root / "pyproject.toml").write_text("[project]\n", encoding="utf-8")
live = root / "venv"
_python(live).parent.mkdir(parents=True)
_python(live).write_bytes(b"live")
(live / "pyvenv.cfg").write_text("home = old-runtime\n", encoding="utf-8")
generation = root / ".hermes-runtime" / "python" / "generation-test"
candidate_python = generation / "python.exe"
candidate_python.parent.mkdir(parents=True)
candidate_python.write_bytes(b"candidate")
candidate = root / ".hermes-runtime" / "venv-candidate-test"
_python(candidate).parent.mkdir(parents=True)
_python(candidate).write_bytes(b"candidate")
(candidate / "pyvenv.cfg").write_text(
f"home = {generation}\n",
encoding="utf-8",
)
current = SQLiteRuntimeInfo(
executable=_python(live),
base_prefix=live,
python_version=(3, 11, 15),
sqlite_version=(3, 50, 4),
sqlite_version_string="3.50.4",
sqlite_source_id="old",
)
fixed = SQLiteRuntimeInfo(
executable=candidate_python,
base_prefix=generation,
python_version=(3, 11, 15),
sqlite_version=(3, 53, 1),
sqlite_version_string="3.53.1",
sqlite_source_id="new",
)
with (
patch(
"hermes_cli.managed_uv.probe_sqlite_runtime",
side_effect=[current, current],
),
patch(
"hermes_cli.managed_uv._windows_runtime_holders",
return_value=(False, ""),
),
patch(
"hermes_cli.managed_uv._install_safe_python_generation",
return_value=(generation, candidate_python, fixed),
),
patch(
"hermes_cli.managed_uv._stage_candidate_venv",
return_value=candidate,
),
patch(
"hermes_cli.managed_uv._cut_over_windows_runtime_config",
return_value=(True, True, fixed, ""),
) as windows_cutover,
patch("hermes_cli.managed_uv._cut_over_candidate") as directory_cutover,
):
result = repair_vulnerable_runtime("uv", project_root=root)
assert result.status == "repaired"
assert result.sqlite_before == "3.50.4"
assert result.sqlite_after == "3.53.1"
windows_cutover.assert_called_once_with(candidate, live=live)
directory_cutover.assert_not_called()
assert not candidate.exists()
assert generation.exists()