ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(cli-misc): macos_tcc_anchor.py — shared managed-venv/marker/staging helpers, module-level name tuples

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:10:56 -07:00
parent bf7cd6a6a6
commit a4ab7170f4
1 changed files with 109 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/macos_tcc_anchor.py
+109 -141
View File
@@ -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)