"""User-local tools. No sudo and no changes to the application's interpreter.""" from __future__ import annotations from collections import deque from collections.abc import Callable import json import os from pathlib import Path import selectors import shutil import signal import subprocess import sys import time Progress = Callable[[str, float | None, float | None], None] def data_dir() -> Path: """The VoiceForge data/prefix directory. VOICEFORGE_HOME wins (the single-directory launcher exports it), then XDG_DATA_HOME/voiceforge, then ~/.local/share/voiceforge. A relative VOICEFORGE_HOME resolves against the current directory. """ home = os.environ.get("VOICEFORGE_HOME") if home: return Path(home).absolute() root = Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share"))) if not root.is_absolute(): root = Path.home() / ".local/share" return root / "voiceforge" def tool_env() -> dict[str, str]: """Sanitized environment for bootstrap, setup, and worker subprocesses. Host Python settings are removed so a virtualenv or sitecustomize on the machine cannot leak into managed environments. Installation-target, index, and constraint overrides are dropped and package-manager configuration files disabled, so packages can only come from the explicitly passed trusted indexes and land inside the data directory; every cache (pip, uv, managed Python, XDG, temporary files) stays there too. """ env = os.environ.copy() for name in ("PYTHONPATH", "PYTHONHOME", "VIRTUAL_ENV", "PIP_TARGET", "PIP_PREFIX", "PIP_USER", "PIP_FIND_LINKS", "PIP_INDEX_URL", "PIP_EXTRA_INDEX_URL", "PIP_CONSTRAINT", "UV_INDEX_URL", "UV_DEFAULT_INDEX", "UV_EXTRA_INDEX_URL", "UV_INDEX", "UV_FIND_LINKS", "UV_CONSTRAINT"): env.pop(name, None) env["PYTHONNOUSERSITE"] = "1" env["PIP_CONFIG_FILE"] = os.devnull env["UV_NO_CONFIG"] = "1" cache = data_dir() / "cache" env["XDG_CACHE_HOME"] = str(cache) env["PIP_CACHE_DIR"] = str(cache / "pip") env["UV_CACHE_DIR"] = str(cache / "uv") env["UV_PYTHON_INSTALL_DIR"] = str(data_dir() / "uv/python") tmp = data_dir() / "tmp" try: tmp.mkdir(parents=True, exist_ok=True) env["TMPDIR"] = str(tmp) except OSError: pass # Best effort: never break subprocesses over temporary-file containment. return env def terminate_group(process: subprocess.Popen) -> None: """SIGTERM the child's whole process group, escalate to SIGKILL, and reap. The group is signalled even when the leader has already exited, because descendants can outlive it while still holding the output pipe open. A race between exit and signalling is tolerated. """ for sig in (signal.SIGTERM, signal.SIGKILL): try: os.killpg(process.pid, sig) except ProcessLookupError: pass try: process.wait(timeout=5) except subprocess.TimeoutExpired: continue try: os.killpg(process.pid, 0) except ProcessLookupError: return process.wait() def run_process(args: list[str], stage: str, progress: Progress | None = None, *, env: dict[str, str] | None = None) -> None: """Drain output without blocking; send heartbeat callbacks every 0.25s. Worker JSON records carry frame/byte counts. Other processes report elapsed seconds as completed with total=None. Only the last 64 KiB of diagnostics are retained. Callback exceptions (including cancellation) terminate the process group. """ started = time.monotonic() current_stage, completed, total = stage, None, None diagnostics: deque[bytes] = deque(maxlen=16) pending = b"" if progress: progress(stage, None, None) with subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, start_new_session=True) as process: assert process.stdout is not None os.set_blocking(process.stdout.fileno(), False) try: with selectors.DefaultSelector() as selector: selector.register(process.stdout, selectors.EVENT_READ) while selector.get_map() or process.poll() is None: for key, _ in selector.select(timeout=0.25): chunk = os.read(key.fd, 4096) if not chunk: selector.unregister(key.fileobj) continue diagnostics.append(chunk) pending += chunk while b"\n" in pending: line, pending = pending.split(b"\n", 1) if line.startswith(b"VOICEFORGE_PROGRESS "): event = json.loads(line[len(b"VOICEFORGE_PROGRESS "):]) current_stage, completed, total = event if len(pending) > 65536: pending = pending[-4096:] if progress: progress(current_stage, completed if total is not None else time.monotonic() - started, total) code = process.wait() if code: detail = b"".join(diagnostics).decode("utf-8", errors="replace") raise RuntimeError(f"{stage} failed (exit {code}):\n{detail}") except BaseException: terminate_group(process) raise def _uv_supports_relocatable(uv: str) -> bool: """Whether this uv understands the venv flags AI setup relies on. A host uv can be older than the pinned one; capability is checked directly instead of trusting the version string. """ try: result = subprocess.run([uv, "venv", "--help"], capture_output=True, text=True, check=True) except (OSError, subprocess.SubprocessError): return False return "--relocatable" in result.stdout + result.stderr def ensure_uv(progress: Progress | None = None) -> str: """Return a uv that supports the flags AI setup relies on. The launcher's contained bootstrap uv is preferred; a host uv is trusted only if it actually supports --relocatable. Otherwise uv==0.8.17 is installed from PyPI (trust PyPI's distribution, not a downloaded shell script). """ contained = data_dir() / "bootstrap/bin/uv" if contained.is_file() and os.access(contained, os.X_OK): return str(contained) found = shutil.which("uv") if found and _uv_supports_relocatable(found): return found root = contained.parent.parent root.parent.mkdir(parents=True, exist_ok=True) python = sys.executable or shutil.which("python3") env = tool_env() run_process([python, "-m", "venv", str(root)], "Creating uv bootstrap", progress, env=env) run_process([str(root / "bin/python"), "-m", "pip", "install", "--index-url", "https://pypi.org/simple", "uv==0.8.17"], "Installing uv from PyPI", progress, env=env) if not (contained.is_file() and os.access(contained, os.X_OK)): raise RuntimeError("Installing uv from PyPI did not produce a usable " "uv; check for host pip settings (such as " "PIP_TARGET) that redirect installations outside " "the data directory.") return str(contained) def ensure_ffmpeg() -> str: """Return the bundled FFmpeg, or a system one where the bundle is absent, not ffprobe. The launcher installs imageio-ffmpeg via the bundled-ffmpeg extra; direct package installs should declare it as a dependency. This helper never runs pip in the current interpreter. """ try: import imageio_ffmpeg return imageio_ffmpeg.get_ffmpeg_exe() except (ImportError, RuntimeError) as error: found = shutil.which("ffmpeg") if found: return found raise RuntimeError("FFmpeg is unavailable. Rebuild the contained " "runtime with './producer.sh --rebuild', or " "install a system FFmpeg.") from error