From a4ab7170f46cf82ec1a7ca730f74172ff032f187 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:10:56 -0700 Subject: [PATCH] =?UTF-8?q?refactor(cli-misc):=20macos=5Ftcc=5Fanchor.py?= =?UTF-8?q?=20=E2=80=94=20shared=20managed-venv/marker/staging=20helpers,?= =?UTF-8?q?=20module-level=20name=20tuples?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/macos_tcc_anchor.py | 250 ++++++++++++++------------------- 1 file changed, 109 insertions(+), 141 deletions(-) diff --git a/hermes_cli/macos_tcc_anchor.py b/hermes_cli/macos_tcc_anchor.py index cc08a113ca..985f67da1f 100644 --- a/hermes_cli/macos_tcc_anchor.py +++ b/hermes_cli/macos_tcc_anchor.py @@ -1,11 +1,12 @@ -"""Stable macOS TCC anchor for the uv-managed Python interpreter (#95596). +"""Stable macOS TCC anchor for the uv-managed Python interpreter. -1. Aliases are materialized as real-file copies of the anchor, never symlinks. 2. If the store ships -``libpython*``, it is hardlinked into ``venv/lib/`` (copy if the store is on another device). -Existing ``LC_RPATH`` already points at ``@executable_path/../lib`` — no rewrite. 3. - -All functions are no-ops on non-macOS and for interpreters that are not uv-managed. Best-effort: -never raises to callers. +macOS TCC grants are keyed to the interpreter binary's path; a venv ``bin/python`` that symlinks +into uv's store changes identity on every interpreter upgrade. The anchor replaces it with a +signed real-file copy, gated on a real boot, with uv's alias names (``python3``, +``python3.N``) materialized as real-file copies too (never symlinks — the #95541 crash shape) +and the store's ``libpython*`` hardlinked into ``venv/lib/`` (existing ``LC_RPATH`` already points +at ``@executable_path/../lib``). All functions are no-ops off macOS and for non-uv interpreters, +and never raise to callers. """ from __future__ import annotations @@ -17,6 +18,7 @@ import platform import re import shutil import subprocess +import sys import tempfile from pathlib import Path @@ -29,19 +31,20 @@ logger = logging.getLogger(__name__) _MARKER_NAME = ".tcc-anchor-source" _STORE_COMMON_MARKERS = ("cpython-", "-macos-") -# The runtime-store marker is derived from managed_uv so a rename of the -# repair-generation directory cannot silently stop the anchor from matching. +# Derived from managed_uv so a rename of the repair-generation directory cannot silently stop +# the anchor from matching. _STORE_ROOT_MARKERS = ("/uv/python/", f"/{_RUNTIME_DIR_NAME}/python/") +_ALIAS_NAMES = ("python3", f"python3.{sys.version_info.minor}") +_STORE_BIN_NAMES = (f"python3.{sys.version_info.minor}", "python3", "python") + class _BootGateFailed(Exception): """Staged copy refused to boot; the live venv must stay untouched.""" def _marker_value(source_file: Path) -> str: - """Canonical marker value: fully resolved so symlinked spellings of the same store binary - (``cpython-3.11-macos-*`` → ``cpython-3.11.15-macos-*``) compare equal. - """ + """Fully resolved so symlinked spellings of the same store binary compare equal.""" return os.path.realpath(str(source_file)) @@ -49,20 +52,6 @@ def is_macos() -> bool: return platform.system() == "Darwin" -def _sibling_names() -> tuple[str, ...]: - """Alias names uv creates inside the venv bin dir.""" - import sys as _sys - - return ("python3", f"python3.{_sys.version_info.minor}") - - -def _store_bin_names() -> tuple[str, ...]: - """Preferred interpreter file names inside a store ``bin`` dir.""" - import sys as _sys - - return (f"python3.{_sys.version_info.minor}", "python3", "python") - - def _is_uv_macos_store(path: str) -> bool: normalized = path.replace("\\", "/") if not all(marker in normalized for marker in _STORE_COMMON_MARKERS): @@ -71,11 +60,7 @@ def _is_uv_macos_store(path: str) -> bool: def _venv_dir(project_root: Path | None = None) -> Path | None: - root = ( - Path(project_root) - if project_root is not None - else Path(__file__).resolve().parents[1] - ) + root = Path(project_root) if project_root is not None else Path(__file__).resolve().parents[1] for name in ("venv", ".venv"): candidate = root / name venv_py = venv_python_path(candidate) @@ -91,7 +76,7 @@ def _interpreter_file(src: str | Path) -> Path | None: return p if not p.is_dir(): return None - for name in _store_bin_names(): + for name in _STORE_BIN_NAMES: candidate = p / name if candidate.is_file(): return candidate @@ -105,14 +90,13 @@ def _interpreter_file(src: str | Path) -> Path | None: def _interpreter_source(venv_dir: Path) -> str | None: - """Return the interpreter file the venv currently resolves to.""" + """Return the interpreter file the venv currently resolves to (symlink target or pyvenv.cfg home).""" venv_py = venv_python_path(venv_dir) if venv_py.is_symlink(): try: - resolved = venv_py.resolve(strict=False) + return str(venv_py.resolve(strict=False)) except OSError: return None - return str(resolved) cfg = venv_dir / "pyvenv.cfg" if not cfg.is_file(): return None @@ -120,8 +104,7 @@ def _interpreter_source(venv_dir: Path) -> str | None: try: for line in cfg.read_text(encoding="utf-8").splitlines(): if line.lower().startswith("home"): - _, _, home = line.partition("=") - home = home.strip() + home = line.partition("=")[2].strip() break except OSError: return None @@ -131,17 +114,34 @@ def _interpreter_source(venv_dir: Path) -> str | None: return str(interp) if interp is not None else None +def _managed_venv(project_root: Path | None) -> tuple[Path, Path, str] | str: + """``(venv_dir, venv_py, source)`` for a uv-managed macOS venv, else the skip reason.""" + if not is_macos(): + return "not macOS" + venv_dir = _venv_dir(project_root) + if venv_dir is None: + return "no venv interpreter" + source = _interpreter_source(venv_dir) + if source is None or not _is_uv_macos_store(source): + return "interpreter not uv-managed (stable path)" + return venv_dir, venv_python_path(venv_dir), source + + def _anchor_marker(venv_bin: Path) -> Path: return venv_bin / _MARKER_NAME -def _write_marker(venv_bin: Path, source_file: Path) -> None: - """Write the anchor marker atomically via the shared helper. +def _marker_matches(venv_bin: Path, expected: str) -> bool: + marker = _anchor_marker(venv_bin) + try: + return marker.is_file() and marker.read_text(encoding="utf-8").strip() == expected + except OSError: + return False - A concurrent ensure (update + doctor --fix) must never observe a partially-written marker: a - torn read would compare unequal and trigger a spurious reinstall, and ``write_text`` alone is - not atomic. - """ + +def _write_marker(venv_bin: Path, source_file: Path) -> None: + """Atomic: a concurrent ensure (update + doctor --fix) must never read a torn marker, which + would compare unequal and trigger a spurious reinstall.""" atomic_write_text( _anchor_marker(venv_bin), _marker_value(source_file), @@ -154,13 +154,11 @@ def _store_root(source_file: Path) -> Path: return source_file.resolve(strict=False).parent.parent -def _provision_libpython( - venv_dir: Path, source_file: Path, *, refresh: bool = False -) -> None: +def _provision_libpython(venv_dir: Path, source_file: Path, *, refresh: bool = False) -> None: """Hardlink (else copy) store ``libpython*`` into ``venv/lib/``. Provision-if-present: a surplus hardlink on a statically-linked build is free; a missed - detection is the only way #95425 returns. + detection is the only way the dylib-not-found crash returns. """ src_lib = _store_root(source_file) / "lib" if not src_lib.is_dir(): @@ -190,46 +188,54 @@ def _provision_libpython( logger.debug("libpython provision skipped", exc_info=True) +def _stage_copy(venv_bin: Path, prefix: str, source: Path) -> Path: + """Copy *source* to a unique (mkstemp) executable staging file in *venv_bin*. + + Unique names mean a concurrent ensure cannot promote another run's truncated interim copy. + """ + fd, tmp_name = tempfile.mkstemp(prefix=prefix, dir=str(venv_bin)) + os.close(fd) + tmp_path = Path(tmp_name) + try: + shutil.copy2(source, tmp_path) + os.chmod(tmp_path, source.stat().st_mode | 0o111) + except BaseException: + _discard(tmp_path) + raise + return tmp_path + + +def _discard(tmp_path: Path | None) -> None: + if tmp_path is not None: + try: + tmp_path.unlink(missing_ok=True) + except OSError: + pass + + def _copy_alias(venv_bin: Path, name: str, anchor: Path) -> bool: """Materialize *name* as a real-file copy of *anchor* (atomic rename). Returns False (and warns) on failure: a leftover alias *symlink* to the anchor is the exact - #95541 crash shape, so callers must know when the alias set is incomplete. The staging name is - unique (mkstemp) so a concurrent ensure (update + doctor --fix) cannot promote a truncated - interim copy. + crash shape, so callers must know when the alias set is incomplete. """ tmp_path: Path | None = None try: - fd, tmp_name = tempfile.mkstemp(prefix=f".{name}.tcc-", dir=str(venv_bin)) - os.close(fd) - tmp_path = Path(tmp_name) - shutil.copy2(anchor, tmp_path) - os.chmod(tmp_path, anchor.stat().st_mode | 0o111) + tmp_path = _stage_copy(venv_bin, f".{name}.tcc-", anchor) os.replace(tmp_path, venv_bin / name) return True except OSError as exc: logger.warning("TCC anchor alias %s not materialized: %s", name, exc) - if tmp_path is not None: - try: - tmp_path.unlink(missing_ok=True) - except OSError: - pass + _discard(tmp_path) return False -def _materialize_aliases( - venv_bin: Path, anchor: Path, *, refresh: bool = False -) -> bool: - """Materialize uv alias names as real-file copies of the anchor. - - Returns True only when every alias that needed materializing succeeded. - """ - names = set(_sibling_names()) +def _materialize_aliases(venv_bin: Path, anchor: Path, *, refresh: bool = False) -> bool: + """Materialize uv alias names as real-file copies; True only if every needed one succeeded.""" + names = set(_ALIAS_NAMES) try: names.update( - p.name - for p in venv_bin.glob("python3*") - if re.fullmatch(r"python3(\.\d+)?", p.name) + p.name for p in venv_bin.glob("python3*") if re.fullmatch(r"python3(\.\d+)?", p.name) ) except OSError: pass @@ -241,7 +247,6 @@ def _materialize_aliases( ok = _copy_alias(venv_bin, name, anchor) and ok except OSError: ok = False - continue return ok @@ -256,8 +261,7 @@ def _passes_boot_gate(staged: Path, venv_dir: Path) -> bool: env = { k: v for k, v in os.environ.items() - if k not in ("PYTHONHOME", "PYTHONPATH", "PYTHONSTARTUP", - "__PYVENV_LAUNCHER__") + if k not in ("PYTHONHOME", "PYTHONPATH", "PYTHONSTARTUP", "__PYVENV_LAUNCHER__") } try: proc = subprocess.run( @@ -294,12 +298,8 @@ def _install_anchor(venv_dir: Path, source_file: Path) -> None: _provision_libpython(venv_dir, source_file, refresh=True) - fd, tmp_name = tempfile.mkstemp(prefix=".python-tcc-", dir=str(venv_bin)) - os.close(fd) - tmp_path = Path(tmp_name) + tmp_path = _stage_copy(venv_bin, ".python-tcc-", source_file) try: - shutil.copy2(source_file, tmp_path) - os.chmod(tmp_path, source_file.stat().st_mode | 0o111) try: from hermes_cli.managed_uv import _macos_sign_managed_python @@ -307,17 +307,12 @@ def _install_anchor(venv_dir: Path, source_file: Path) -> None: except Exception: # pragma: no cover - never block the anchor logger.debug("anchor copy signing skipped", exc_info=True) if not _passes_boot_gate(tmp_path, venv_dir): - raise _BootGateFailed( - f"staged copy at {tmp_path} failed encodings/prefix probe" - ) + raise _BootGateFailed(f"staged copy at {tmp_path} failed encodings/prefix probe") os.replace(tmp_path, venv_py) - aliases_ok = _materialize_aliases(venv_bin, venv_py, refresh=True) - if aliases_ok: - # Marker last, atomically: it asserts the WHOLE layout (anchor + - # aliases) is complete. A partially-materialized alias set (the - # #95541 crash shape when an alias stays a symlink) must not read - # "active" in doctor — leaving the marker absent makes the next - # ensure retry the install. + if _materialize_aliases(venv_bin, venv_py, refresh=True): + # Marker last, atomically: it asserts the WHOLE layout (anchor + aliases) is complete. + # A partial alias set must not read "active" in doctor; an absent marker makes the + # next ensure retry the install. _write_marker(venv_bin, source_file) else: logger.warning( @@ -325,44 +320,30 @@ def _install_anchor(venv_dir: Path, source_file: Path) -> None: "incomplete; leaving anchor unmarked so the next run retries" ) except Exception: - try: - tmp_path.unlink(missing_ok=True) - except OSError: - pass + _discard(tmp_path) raise def ensure_tcc_anchor(project_root: Path | None = None) -> Path | None: - """Pin a dylib-complete interpreter anchor for macOS TCC (#95596). + """Pin a dylib-complete interpreter anchor for macOS TCC. - No-op (returns None) on non-macOS, when no venv interpreter exists, or when the interpreter is - not uv-managed. Idempotent. Best-effort — returns None (and logs) if the copy or boot-gate - fails; callers must never depend on success. + No-op (None) on non-macOS, without a venv interpreter, or when the interpreter is not + uv-managed. Idempotent. Best-effort — None (and logs) if the copy or boot-gate fails; callers + must never depend on success. """ - if not is_macos(): - return None - venv_dir = _venv_dir(project_root) - if venv_dir is None: - return None - venv_py = venv_python_path(venv_dir) - if not (venv_py.is_file() or venv_py.is_symlink()): - return None - source = _interpreter_source(venv_dir) - if source is None or not _is_uv_macos_store(source): + found = _managed_venv(project_root) + if isinstance(found, str): return None + venv_dir, venv_py, source = found source_file = _interpreter_file(source) if source_file is None: return None - if not venv_py.is_symlink(): - marker = _anchor_marker(venv_py.parent) + if not venv_py.is_symlink() and _marker_matches(venv_py.parent, _marker_value(source_file)): try: - if marker.is_file() and marker.read_text(encoding="utf-8").strip() == ( - _marker_value(source_file) - ): - _provision_libpython(venv_dir, source_file, refresh=False) - if _passes_boot_gate(venv_py, venv_dir): - _materialize_aliases(venv_py.parent, venv_py) - return venv_py + _provision_libpython(venv_dir, source_file, refresh=False) + if _passes_boot_gate(venv_py, venv_dir): + _materialize_aliases(venv_py.parent, venv_py) + return venv_py except OSError: pass try: @@ -379,29 +360,16 @@ def ensure_tcc_anchor(project_root: Path | None = None) -> Path | None: def tcc_anchor_state(project_root: Path | None = None) -> tuple[str, str]: """Report the anchor state for ``hermes doctor`` as ``(status, detail)``. - ``skip`` = not applicable (non-macOS, no venv, not uv-managed); ``active`` = pinned at a - stable real-file anchor; ``stale`` = pinned but the interpreter changed since the last copy; - ``missing`` = uv-managed interpreter with no anchor installed. + ``skip`` = not applicable; ``active`` = pinned at a stable real-file anchor; ``stale`` = pinned + but the interpreter changed since the last copy; ``missing`` = uv-managed with no anchor. """ - if not is_macos(): - return "skip", "not macOS" - venv_dir = _venv_dir(project_root) - if venv_dir is None: - return "skip", "no venv interpreter" - venv_py = venv_python_path(venv_dir) - if not (venv_py.is_file() or venv_py.is_symlink()): - return "skip", "no venv interpreter" - source = _interpreter_source(venv_dir) - if source is None or not _is_uv_macos_store(source): - return "skip", "interpreter not uv-managed (stable path)" - if not venv_py.is_symlink(): - marker = _anchor_marker(venv_py.parent) - source_file = _interpreter_file(source) - expected = _marker_value(source_file) if source_file is not None else source - try: - if marker.is_file() and marker.read_text(encoding="utf-8").strip() == expected: - return "active", str(venv_py) - except OSError: - pass - return "stale", str(venv_py) - return "missing", str(venv_py) + found = _managed_venv(project_root) + if isinstance(found, str): + return "skip", found + _venv, venv_py, source = found + if venv_py.is_symlink(): + return "missing", str(venv_py) + source_file = _interpreter_file(source) + expected = _marker_value(source_file) if source_file is not None else source + status = "active" if _marker_matches(venv_py.parent, expected) else "stale" + return status, str(venv_py)