fdf9cebe2c
The previous commit dropped --verbose and broke the contract that native build output streams live (tests/pm/test_environment_build.py). uv only forwards backend output at DEBUG level, so keep --verbose and scope it with RUST_LOG=uv_build_frontend=debug: the backend's own lines stream, the interpreter/cache/reflink internals stay silent (2 lines on a no-op sync instead of ~200 on both uv 0.11 and the pinned 0.12.3).
392 lines
18 KiB
Python
392 lines
18 KiB
Python
"""PM's private Python engine: tool routing, isolation and command execution.
|
|
|
|
Operations own source, destination and publication. Bootstrap may inject its
|
|
staged toolchain directly without recursing through the worker it is building.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import codecs
|
|
from collections.abc import Iterator, Mapping, Sequence
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from typing import TextIO
|
|
|
|
from pm.package import InstallError
|
|
|
|
# Deliberately narrow: a fetch timeout or index outage must not be misread as a
|
|
# conflict — and regardless of classification, nothing here ever disables a
|
|
# plugin; the caller decides. Lives here (stdlib-only imports) because the
|
|
# bootstrap runner streams uv output from a pre-3.11 system python where
|
|
# pm.workspace's tomllib import cannot load.
|
|
_RESOLVER_MARKERS = (
|
|
"no solution found",
|
|
"conflicting requirements",
|
|
"conflicting urls",
|
|
"because only the following versions",
|
|
"and your pyproject depends on",
|
|
)
|
|
|
|
|
|
class ResolutionConflict(InstallError):
|
|
"""uv's resolver proved the union has no valid solution."""
|
|
|
|
|
|
def classify_uv_failure(stage: str, returncode: int, output: str) -> InstallError:
|
|
"""Turn a failed `uv <stage>` into the right classified error.
|
|
|
|
Resolver-conflict output → ResolutionConflict; anything else (fetch,
|
|
build, tooling) → plain InstallError with the tail of the output.
|
|
"""
|
|
cause = f"uv {stage} exited {returncode}: {output.strip()[-600:]}"
|
|
lowered = output.lower()
|
|
if any(marker in lowered for marker in _RESOLVER_MARKERS):
|
|
return ResolutionConflict("venv", cause)
|
|
return InstallError("venv", cause)
|
|
|
|
|
|
def _project_name(source: Path) -> str:
|
|
"""``[project].name`` of *source*'s pyproject, on any Python that can run the bootstrap.
|
|
|
|
tomllib is 3.11+; the bootstrap runner stages PM's runtime from a system python that may
|
|
be older, so the one field this module reads is parsed with the stdlib parser when it
|
|
exists and a table-scoped regex otherwise.
|
|
"""
|
|
text = (source / "pyproject.toml").read_text(encoding="utf-8-sig")
|
|
try:
|
|
import tomllib
|
|
except ModuleNotFoundError:
|
|
import re
|
|
table = re.search(r"(?ms)^\[project\]\s*$(.*?)(?=^\[)", text + "\n[")
|
|
match = table and re.search(r'(?m)^name\s*=\s*"([^"]+)"', table.group(1))
|
|
if not match:
|
|
raise InstallError("venv", f"{source / 'pyproject.toml'} has no [project].name")
|
|
return match.group(1)
|
|
return tomllib.loads(text)["project"]["name"]
|
|
|
|
|
|
def prune_site_pth(venv_dir: Path) -> None:
|
|
"""Drop .pth files that must never execute inside a shipped payload.
|
|
|
|
``uv sync`` leaves two behind: ``_virtualenv.pth`` (repoints
|
|
``sys.prefix`` at the venv) and the project's ``__editable__`` pointer
|
|
(names the BUILD machine — the payload wires the repo snapshot itself,
|
|
see scripts/build/launcher_wrapper.py). Bundled launchers process every
|
|
other .pth with ``site.addsitedir()``; pywin32.pth is load-bearing on
|
|
Windows (win32\\lib on sys.path is what makes ``import pywintypes``
|
|
resolve, which portalocker/concurrent-log-handler need to write logs).
|
|
"""
|
|
if (venv_dir / "Scripts").is_dir():
|
|
sites = [venv_dir / "Lib" / "site-packages"]
|
|
else:
|
|
lib = venv_dir / "lib"
|
|
sites = sorted(lib.glob("python*/site-packages")) if lib.is_dir() else []
|
|
for site_dir in sites:
|
|
if not site_dir.is_dir():
|
|
continue
|
|
for pth in site_dir.glob("*.pth"):
|
|
if pth.name == "_virtualenv.pth" or pth.name.startswith("__editable__"):
|
|
try:
|
|
pth.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _read_pipe(fd: int) -> bytes:
|
|
size = 65536
|
|
if sys.platform == "win32":
|
|
import _winapi
|
|
import msvcrt
|
|
|
|
# Bootstrap can run on Python 3.11, before Windows os.set_blocking
|
|
# exists. Peek keeps the sole reader bounded without changing pipe mode.
|
|
try:
|
|
available, _ = _winapi.PeekNamedPipe(msvcrt.get_osfhandle(fd), 0)
|
|
except BrokenPipeError:
|
|
return b""
|
|
if not available:
|
|
raise BlockingIOError
|
|
size = min(size, available)
|
|
return os.read(fd, size)
|
|
|
|
|
|
def _run_streaming(command: list[str], *, cwd: Path, env: dict[str, str],
|
|
timeout: int, output: TextIO) -> subprocess.CompletedProcess:
|
|
"""Keep CI progress live, a bounded diagnostic tail, and a wall-clock timeout."""
|
|
deadline = time.monotonic() + timeout
|
|
proc = subprocess.Popen(command, cwd=str(cwd), env=env, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=0)
|
|
pipe = proc.stdout
|
|
assert isinstance(pipe, io.TextIOWrapper) # Popen was given stdout=PIPE and text=True.
|
|
tail = ""
|
|
conflict = ""
|
|
try:
|
|
# A descendant can keep stdout open after proc exits. Nonblocking reads
|
|
# bound that drain without leaving a thread stuck in readline()/close().
|
|
if sys.platform != "win32":
|
|
os.set_blocking(pipe.fileno(), False)
|
|
decoder = io.IncrementalNewlineDecoder(
|
|
codecs.getincrementaldecoder(pipe.encoding)(errors="replace"), translate=True,
|
|
)
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise subprocess.TimeoutExpired(command, timeout, stderr=tail)
|
|
try:
|
|
data = _read_pipe(pipe.fileno())
|
|
except BlockingIOError:
|
|
time.sleep(min(.05, remaining))
|
|
continue
|
|
text = decoder.decode(data, final=not data)
|
|
if text:
|
|
# Preserve an observed resolver marker even after verbose output
|
|
# evicts it. Scan across read boundaries, never retain the full log.
|
|
if not conflict:
|
|
lowered = (tail + text).lower()
|
|
conflict = next((marker for marker in _RESOLVER_MARKERS if marker in lowered), "")
|
|
tail = (tail + text)[-2000:]
|
|
output.write(text)
|
|
output.flush()
|
|
if not data:
|
|
break
|
|
# EOF can precede process exit; it does not grant another timeout budget.
|
|
try:
|
|
code = proc.wait(timeout=max(0, deadline - time.monotonic()))
|
|
except subprocess.TimeoutExpired:
|
|
raise subprocess.TimeoutExpired(command, timeout, stderr=tail) from None
|
|
except BaseException:
|
|
proc.kill()
|
|
proc.wait(timeout=5)
|
|
raise
|
|
finally:
|
|
pipe.close()
|
|
if conflict and conflict not in tail.lower():
|
|
tail = conflict + "\n" + tail[-(2000 - len(conflict) - 1):]
|
|
return subprocess.CompletedProcess(command, code, "", tail)
|
|
|
|
|
|
def _base_environment(env: Mapping[str, str] | None = None) -> dict[str, str]:
|
|
"""Ambient UV settings never select the project, interpreter or cache.
|
|
|
|
Index and transport settings are the exception (pm.index_config): without
|
|
them mirrored and air-gapped networks cannot resolve anything.
|
|
"""
|
|
from pm.index_config import bridged_index_settings, is_forwarded
|
|
|
|
source = os.environ if env is None else env
|
|
base = {key: value for key, value in source.items()
|
|
if not key.startswith("PYTHON") and key != "VIRTUAL_ENV"
|
|
and (not key.startswith("UV_") or is_forwarded(key))}
|
|
if env is None:
|
|
base.update(bridged_index_settings(os.environ))
|
|
return base
|
|
|
|
|
|
def managed_environment(destination: Path, *, python: Path | None = None,
|
|
cache: Path | None = None, env: Mapping[str, str] | None = None,
|
|
offline: bool = False, explicit: bool = False,
|
|
output: TextIO | None = None, realize: bool = True) -> PythonEnvironment:
|
|
from pm._uv import _toolchain
|
|
from pm.packages import uv_cache_dir
|
|
|
|
tools = _toolchain(explicit=explicit, realize=realize)
|
|
if tools is None:
|
|
raise InstallError("venv", "PM's pinned toolchain is unavailable")
|
|
uv, pinned_python = tools
|
|
python = pinned_python if python is None else python.absolute()
|
|
build_env = _base_environment(env)
|
|
if sys.platform == "darwin" and python.resolve() == pinned_python.resolve():
|
|
# PBS's AR still names its deleted build directory. Its CC is already
|
|
# clang; only the archiver needs a default, and only for our interpreter.
|
|
build_env.setdefault("AR", "/usr/bin/ar")
|
|
return PythonEnvironment(
|
|
uv=uv, python=python,
|
|
destination=destination.absolute(), cache=uv_cache_dir() if cache is None else cache.absolute(),
|
|
env=build_env, offline=offline, output=output,
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def _fresh_build(environment: PythonEnvironment, *, sealed: bool) -> Iterator[None]:
|
|
"""Own only the output claimed by this build, including failed validation."""
|
|
out = environment.destination
|
|
# A concurrent creator wins intact: never enter cleanup before mkdir succeeds.
|
|
out.mkdir(parents=True)
|
|
try:
|
|
environment.create()
|
|
yield
|
|
environment.check()
|
|
if sealed:
|
|
prune_site_pth(out)
|
|
except BaseException:
|
|
shutil.rmtree(out, ignore_errors=True)
|
|
raise
|
|
|
|
|
|
@dataclass(frozen=True, kw_only=True)
|
|
class PythonEnvironment:
|
|
uv: Path
|
|
python: Path
|
|
destination: Path
|
|
cache: Path
|
|
env: Mapping[str, str]
|
|
offline: bool = False
|
|
output: TextIO | None = None
|
|
no_config: bool = False
|
|
|
|
@property
|
|
def executable(self) -> Path:
|
|
from pm.environments import venv_python
|
|
|
|
return venv_python(self.destination)
|
|
|
|
def _run(self, args: list[str], *, cwd: Path, timeout: int) -> subprocess.CompletedProcess:
|
|
# Explicit index credentials survive, but cannot redirect the project,
|
|
# interpreter or cache selected by the operation.
|
|
env = _base_environment(self.env)
|
|
env.update(UV_PYTHON=str(self.python), UV_PROJECT_ENVIRONMENT=str(self.destination),
|
|
UV_CACHE_DIR=str(self.cache), UV_PYTHON_DOWNLOADS="never")
|
|
with tempfile.TemporaryDirectory(prefix="pm-uv-config-") as config:
|
|
env.update(XDG_CONFIG_HOME=config, XDG_CONFIG_DIRS=config)
|
|
command = [str(self.uv), *args]
|
|
if self.no_config and "--no-config" not in command:
|
|
command.append("--no-config")
|
|
if self.offline:
|
|
command.append("--offline")
|
|
try:
|
|
if self.output is not None:
|
|
# uv hides build-backend output until failure without verbose mode,
|
|
# but --verbose alone is uv's DEBUG level: ~200 lines of interpreter
|
|
# and cache internals on every streamed run. RUST_LOG scopes it to the
|
|
# build frontend, so only the backend's own lines reach the user.
|
|
command.append("--verbose")
|
|
env.setdefault("RUST_LOG", "uv_build_frontend=debug")
|
|
return _run_streaming(command, cwd=cwd, env=env, timeout=timeout, output=self.output)
|
|
return subprocess.run(command, cwd=str(cwd), env=env, capture_output=True,
|
|
text=True, encoding="utf-8", errors="replace", timeout=timeout)
|
|
except subprocess.TimeoutExpired as exc:
|
|
from pm.index_config import TIMEOUT_HINT
|
|
|
|
# A silent stall against an unreachable index is the #95608 shape;
|
|
# name the mirror knobs instead of surfacing a raw TimeoutExpired.
|
|
raise InstallError("venv", f"uv {args[0]} timed out after {timeout}s", TIMEOUT_HINT) from exc
|
|
|
|
def create(self) -> None:
|
|
"""Create at the final destination; callers must not move a live venv."""
|
|
self.destination.parent.mkdir(parents=True, exist_ok=True)
|
|
result = self._run(
|
|
["venv", "--relocatable", "--no-project", "--no-config",
|
|
"--python", str(self.python), str(self.destination)],
|
|
cwd=self.destination.parent, timeout=120,
|
|
)
|
|
if result.returncode:
|
|
raise InstallError("venv", f"uv venv failed: {result.stderr[-600:]}")
|
|
|
|
def lock(self, source: Path, *, upgrade: bool = False, timeout: int = 1800) -> None:
|
|
command = ["lock", "--python", str(self.python)]
|
|
if upgrade:
|
|
command.append("--upgrade")
|
|
result = self._run(command, cwd=source, timeout=timeout)
|
|
if result.returncode:
|
|
raise classify_uv_failure("lock", result.returncode, result.stderr or result.stdout)
|
|
|
|
def check_lock(self, source: Path) -> None:
|
|
result = self._run(["lock", "--check", "--python", str(self.python)],
|
|
cwd=source, timeout=1800)
|
|
if result.returncode:
|
|
raise classify_uv_failure("lock", result.returncode, result.stderr or result.stdout)
|
|
|
|
def sync(self, source: Path, *, extras: Sequence[str] = (), groups: Sequence[str] = (),
|
|
timeout: int = 1800, frozen: bool = True, all_extras: bool = False,
|
|
no_install_project: bool = False, locked: bool = False,
|
|
no_default_groups: bool = False, only_groups: bool = False) -> None:
|
|
"""Install the root and every member; resolve only in a writable workspace.
|
|
|
|
``frozen=False`` is reserved for the caller-owned generated workspace,
|
|
never the original project's lock. Seed/replay policy belongs to PM.
|
|
"""
|
|
if only_groups and (not groups or extras or all_extras):
|
|
raise ValueError("group-only builds require groups and cannot select extras")
|
|
if not frozen:
|
|
self.lock(source, timeout=timeout)
|
|
# Locking members alone is insufficient: plain sync only installs root deps.
|
|
command = ["sync", "--locked" if locked else "--frozen", "--all-packages",
|
|
"--python", str(self.python)]
|
|
if no_default_groups:
|
|
command.append("--no-default-groups")
|
|
if all_extras:
|
|
command.append("--all-extras")
|
|
if no_install_project:
|
|
# --all-packages has no single selected project in uv, so
|
|
# --no-install-project alone does not exclude the root. Name it
|
|
# explicitly without dropping member dependencies or installations.
|
|
command += ["--no-install-project", "--no-install-package", _project_name(source)]
|
|
for extra in sorted(set(extras)):
|
|
command += ["--extra", extra]
|
|
for group in sorted(set(groups)):
|
|
command += ["--only-group" if only_groups else "--group", group]
|
|
result = self._run(command, cwd=source, timeout=timeout)
|
|
if result.returncode:
|
|
raise classify_uv_failure("sync", result.returncode, result.stderr or result.stdout)
|
|
|
|
def export_requirements(self, source: Path, out: Path, *, extras: Sequence[str] = (),
|
|
timeout: int = 1800) -> None:
|
|
command = ["export", "--frozen", "--python", str(self.python), "--no-default-groups",
|
|
"--no-emit-project", "--no-hashes", "--no-annotate", "--no-header",
|
|
"--format", "requirements-txt", "--output-file", str(out)]
|
|
for extra in sorted(set(extras)):
|
|
command += ["--extra", extra]
|
|
result = self._run(command, cwd=source, timeout=timeout)
|
|
if result.returncode:
|
|
raise classify_uv_failure("export", result.returncode, result.stderr or result.stdout)
|
|
|
|
def install_requirements(self, requirements: Sequence[str], *, wheelhouse: Path | None = None) -> None:
|
|
if not requirements:
|
|
return
|
|
# A file avoids command-line length limits and shell/marker quoting.
|
|
with tempfile.TemporaryDirectory(prefix="pm-requirements-") as temporary:
|
|
requirements_file = Path(temporary) / "requirements.txt"
|
|
requirements_file.write_text("\n".join(requirements) + "\n", encoding="utf-8")
|
|
self._install_requirements_file(requirements_file, wheelhouse=wheelhouse)
|
|
|
|
def _install_requirements_file(self, requirements: Path, *, wheelhouse: Path | None = None,
|
|
timeout: int = 1800) -> None:
|
|
command = ["pip", "install", "--no-config", "--python", str(self.executable),
|
|
"--requirements", str(requirements)]
|
|
if wheelhouse is not None:
|
|
command += ["--no-index", "--only-binary", ":all:",
|
|
"--find-links", str(wheelhouse.absolute())]
|
|
result = self._run(command, cwd=requirements.parent, timeout=timeout)
|
|
if result.returncode:
|
|
raise classify_uv_failure("pip", result.returncode, result.stderr or result.stdout)
|
|
|
|
def install_wheelhouse(self, source: Path, wheelhouse: Path, *, timeout: int = 1800) -> None:
|
|
"""Install rebuilt wheels whose hashes the bundle manifest owns, not uv.lock."""
|
|
requirements = source / "requirements.txt"
|
|
self.export_requirements(source, requirements, timeout=timeout)
|
|
self._install_requirements_file(requirements, wheelhouse=wheelhouse, timeout=timeout)
|
|
self.check()
|
|
|
|
def prune_cache(self, *, ci: bool = False) -> None:
|
|
command = ["cache", "prune", "--no-config"]
|
|
if ci:
|
|
command += ["--ci", "--force"]
|
|
result = self._run(command, cwd=Path.cwd(), timeout=1800)
|
|
if result.returncode:
|
|
raise InstallError("uv", f"cache pruning failed: {result.stderr[-600:]}")
|
|
|
|
def check(self) -> None:
|
|
result = self._run(
|
|
["pip", "check", "--no-config", "--python", str(self.executable)],
|
|
cwd=self.destination.parent, timeout=60,
|
|
)
|
|
if result.returncode:
|
|
raise InstallError("venv", f"dependency validation failed: {result.stderr[-600:]}")
|