path: root/app/backends/common.py

"""Shared helpers for the backend setup wizards.

Every TTS backend setup wizard (audio.cpp, qwen, faster) lives in its own
module under ``backends``; this module holds the pieces more than one of
them needs: .wav discovery, path normalization, and the regex edit that
keeps ``app/converter/config.py`` in sync with the choices made in a wizard.
It deliberately imports nothing from the other backend modules (or the
TUI) so it can be reused without pulling curses into a non-interactive
run.
"""

import os
import re
import sys
import time
import urllib.parse
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

import logging_kit

# Messages queued while the TUI is on screen, printed to the real console
# after the curses session ends (see ui.hub.run). Build/setup steps that
# fail inside the TUI record here so the user gets a copy-pastable command
# and a log path once the TUI exits, instead of losing the output.
_POST_TUI_NOTICES: List[str] = []

# The tts-audiobook-generator checkout root (where audiobook.py lives).
# Everything non-user-facing lives under ./app: the source packages
# (backends, converter, ui), the generated dirs (envs, chunks, logs, debug),
# and the backend checkouts (app/audio.cpp, app/faster-qwen3-tts).
TTS_ROOT = Path(__file__).resolve().parent.parent.parent

# The single "everything else" directory under TTS_ROOT.
APP_DIR = TTS_ROOT / "app"

# app/logs — build/server/conversion logs (already gitignored). Naming and
# retention policy lives in logging_kit (streams vs. timestamped artifacts).
LOG_DIR = logging_kit.LOG_DIR

# The project's sample-voice directory: .wav files dropped here are offered
# as the default source when a setup/configure wizard asks for a wav
# directory (both the TUI browser start and the --wavs flag default).
VOICES_DIR = TTS_ROOT / "voices"

# app/converter/config.py — rewritten in place by update_config_value so the
# converter picks up the host/port/voice a wizard configured.
CONFIG_PATH = APP_DIR / "converter" / "config.py"

# Output directory of tts-audiobook-generator; never offered as a .wav
# source by detect_wav_dir.
TTS_OUTPUT_DIR = "output"

# The voice-transcript mapping file audio.cpp reads from its voice_dir.
# (The faster backend uses voices.json instead; see backends.faster.)
PROMPT_TEXT_FILENAME = "prompt_text"


def record_post_tui_notice(text: str) -> None:
    """Queue a message to print to the console after the TUI session ends.

    The TUI runs in a curses session, so ``print`` during it does not reach
    the real terminal. Steps that fail inside the TUI (e.g. the audio.cpp
    build) record a copy-pastable command and a log path here; ``ui.hub.run``
    drains the queue after the session ends.
    """
    _POST_TUI_NOTICES.append(text)


def drain_post_tui_notices() -> List[str]:
    """Return and clear the queued post-TUI messages."""
    notices = list(_POST_TUI_NOTICES)
    _POST_TUI_NOTICES.clear()
    return notices


def cancel_requested(cancel) -> bool:
    """True when CANCEL (a ``threading.Event``) is given and set.

    Shared guard for the multi-phase uninstall actions: cancellation is
    honored only between phases (stop servers / pip / delete files), so a
    phase that already started always runs to completion and an uninstall
    never tears halfway. Callers return 130 when this fires before a
    pending phase.
    """
    return cancel is not None and cancel.is_set()


def normalize_dir_arg(value: str) -> Path:
    """Normalize a user-supplied path argument.

    Strips surrounding quotes (a common copy-paste artifact), expands a
    leading ``~``, and resolves the result to an absolute path so relative
    paths are always validated against the current working directory.
    """
    cleaned = value.strip()
    if len(cleaned) >= 2 and cleaned[0] == cleaned[-1] and cleaned[0] in "\"'":
        cleaned = cleaned[1:-1]
    return Path(os.path.expanduser(cleaned)).resolve()


def resolve_wav_dir_arg(value: str) -> Path:
    """Normalize a user-supplied wav directory argument."""
    return normalize_dir_arg(value)


def find_wav_files(input_dir: Path) -> List[Path]:
    """Return the .wav files in INPUT_DIR, sorted alphabetically by name.

    A missing or unreadable directory yields [] so callers can treat it
    like an empty directory (matching ``count_wavs``).
    """
    try:
        return sorted(
            (path for path in input_dir.iterdir()
             if path.is_file() and path.suffix.lower() == ".wav"),
            key=lambda path: path.name.lower(),
        )
    except OSError:
        return []


def count_wavs(directory: Path) -> int:
    """Count the .wav files in DIRECTORY (0 when it cannot be read)."""
    try:
        return sum(1 for path in directory.iterdir()
                   if path.is_file() and path.suffix.lower() == ".wav")
    except OSError:
        return 0


def detect_wav_dir(audiocpp_dir: Path, tts_root: Path) -> Optional[Path]:
    """Find a unique directory that directly contains .wav files.

    Looks shallowly (the root itself and its immediate subdirectories) in
    both the audio.cpp checkout and the tts-audiobook-generator root (where
    audiobook.py lives), since clone reference .wavs commonly live in
    either. The tts-audiobook-generator ``output/`` directory is excluded.
    When exactly one candidate is found it is returned (as a starting
    directory for the .wav browser); when none or several are found None is
    returned so the caller falls back to its default start location.
    """
    candidates: List[Path] = []
    seen: Set[Path] = set()

    def consider(directory: Path) -> None:
        try:
            resolved = directory.resolve()
        except OSError:
            return
        if resolved in seen:
            return
        seen.add(resolved)
        if count_wavs(directory) > 0:
            candidates.append(directory)

    for root in (audiocpp_dir, tts_root):
        if not root.is_dir():
            continue
        consider(root)
        try:
            children = sorted(root.iterdir(), key=lambda p: p.name.lower())
        except OSError:
            continue
        for child in children:
            if not child.is_dir() or child.name.startswith("."):
                continue
            if root == tts_root and child.name == TTS_OUTPUT_DIR:
                continue
            consider(child)

    if len(candidates) == 1:
        return candidates[0]
    return None


def wav_dir_info(directory: Path) -> Tuple[str, str]:
    """TUI status describing the directory listed in the wav browser."""
    count = count_wavs(directory)
    if count:
        wavs = ".wav" if count == 1 else ".wavs"
        return (f"{count} {wavs} found in this directory. Press Enter.",
                "ok")
    return ("No .wav files found in this directory", "warn")


def wav_dir_preview(directory: Path) -> Tuple[str, str]:
    """TUI status describing a highlighted subdirectory in the wav browser."""
    count = count_wavs(directory)
    if count:
        wavs = ".wav" if count == 1 else ".wavs"
        return (f"{count} {wavs}", "ok")
    return ("no .wav files", "info")


def port_of(url: str, fallback: int) -> int:
    """URL's explicit port, else FALLBACK (invalid URLs fall back too)."""
    try:
        return urllib.parse.urlsplit(url).port or fallback
    except ValueError:
        return fallback


def url_with_port(url: str, port: int) -> str:
    """Return URL with its port replaced/inserted as PORT.

    Preserves the userinfo ("user:pass@host") and brackets IPv6 hosts
    ("[::1]"), which a plain f"{host}:{port}" rebuild would mangle.
    """
    parts = urllib.parse.urlsplit(url)
    host = parts.hostname or "127.0.0.1"
    if ":" in host and not host.startswith("["):
        host = f"[{host}]"
    netloc = f"{host}:{port}"
    if parts.username:
        cred = parts.username
        if parts.password:
            cred = f"{cred}:{parts.password}"
        netloc = f"{cred}@{netloc}"
    return urllib.parse.urlunsplit(
        (parts.scheme or "http", netloc, parts.path, "", ""))


def normalize_remote_url(value: str) -> str:
    """Normalize a user-supplied remote server URL, or '' for "disabled".

    Accepts a bare ``host[:port]`` (a scheme of ``http`` is assumed), a full
    ``http(s)://host[:port][/path]`` URL, or the empty string (no remote
    server configured). Returns the normalized URL (bare host:port becomes
    ``http://host:port``). Raises ValueError for anything else — a missing
    host, a host containing whitespace, or a non-numeric port.
    """
    cleaned = value.strip()
    if not cleaned:
        return ""
    parts = urllib.parse.urlsplit(cleaned)
    if not parts.scheme:
        # Bare host[:port] — add the default scheme so netloc/host/port
        # parse cleanly. An explicit scheme is kept as-is (so "http://"
        # with no host fails the host check below).
        parts = urllib.parse.urlsplit(f"http://{cleaned}")
    host = parts.hostname
    if not host or any(ch.isspace() for ch in host):
        raise ValueError(
            "Enter a host:port (e.g. 10.20.30.40:8000) or a full URL "
            f"(e.g. http://10.20.30.40:8000); got {value!r}")
    try:
        parts.port  # noqa: B018 -- accessing .port raises ValueError when bad
    except ValueError as exc:
        raise ValueError(
            f"Invalid port in remote URL {value!r}: {exc}") from exc
    return urllib.parse.urlunsplit(
        (parts.scheme or "http", parts.netloc, parts.path, "", ""))


def parse_request_options(text: str) -> Dict[str, str]:
    """Parse a user-supplied ``KEY=VALUE`` option string into a dict.

    Items are separated by commas or whitespace; each must contain an
    ``=`` with a non-empty key. Values are kept verbatim (only the key is
    stripped), so e.g. ``speed=1.1`` yields ``{"speed": "1.1"}`` — the
    audio.cpp server coerces per-model option values itself. A blank
    string yields {}. Raises ValueError with a user-facing message when
    an item lacks ``=`` or has an empty key; later duplicates of a key
    override earlier ones.
    """
    options: Dict[str, str] = {}
    for item in text.replace(",", " ").split():
        key, sep, value = item.partition("=")
        if not sep or not key.strip():
            raise ValueError(
                f"Request options expect KEY=VALUE items "
                f"(e.g. emotion=neutral); got {item!r}")
        options[key.strip()] = value
    return options


def server_running(url: str, timeout: float = 0.3) -> bool:
    """True when something accepts TCP connections at URL's host:port.

    A protocol-agnostic socket connect: an HTTP TTS server that is up will
    accept the connection (we do not need to speak HTTP to know it is
    listening). Returns False on any parse or connection error, so a
    misconfigured URL never blocks the hub — it just reports the backend
    as not running. Used by each backend's ``detect()`` to set
    ``BackendStatus.running``.
    """
    import socket
    try:
        parts = urllib.parse.urlsplit(url)
        host = parts.hostname or "127.0.0.1"
        port = parts.port or (443 if (parts.scheme or "http") == "https"
                              else 80)
    except ValueError:
        return False
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def update_config_value(key: str, value,
                        config_path: Optional[Path] = None) -> bool:
    """Set ``KEY`` to VALUE in app/converter/config.py and in memory.

    Only the value of the named assignment changes: indentation and any
    trailing comment are preserved. Strings render double-quoted; other
    literals (ints, booleans) render bare. After a successful write (or
    when the file already holds VALUE) the new value is mirrored onto the
    imported ``converter.config`` module, so a wizard's change takes
    effect immediately instead of only after the next process start.
    Returns True when the file now holds VALUE, False when it could not
    be read or written (or KEY has no line in it).
    """
    path = Path(config_path) if config_path is not None else CONFIG_PATH
    rendered = f'"{value}"' if isinstance(value, str) else str(value)
    try:
        text = path.read_text(encoding="utf-8")
        match = re.search(
            rf'(?m)^(\s*{re.escape(key)}\s*=\s*)("[^"]*"|\S+)(\s*(?:#.*)?)$',
            text)
        if match is None:
            return False
        if match.group(2) != rendered:
            text = text[:match.start(2)] + rendered + text[match.end(2):]
            path.write_text(text, encoding="utf-8")
    except OSError:
        return False
    # Local import: this module must stay importable before the venv
    # exists (backends.envs bootstraps from it), and converter.config is
    # stdlib-only constants, safe to load whenever a wizard runs.
    from converter import config as _config
    setattr(_config, key, value)
    return True


def read_prompt_text(prompt_path: Path) -> Dict[str, str]:
    """Parse a prompt_text file into a stem -> transcript mapping.

    Lines are ``<name>|<transcript>``; blank lines are skipped and a line
    without a ``|`` separator is treated as a name with an empty transcript.
    Returns an empty mapping when the file does not exist.
    """
    if not prompt_path.exists():
        return {}
    mapping: Dict[str, str] = {}
    for line in prompt_path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        if "|" in line:
            name, _, text = line.partition("|")
        else:
            name, text = line, ""
        mapping[name.strip()] = text
    return mapping


def write_prompt_text(wav_dir: Path,
                      transcripts: Dict[str, str]) -> Path:
    """Write the voice_dir prompt_text mapping into WAV_DIR.

    One ``<basename-without-extension>|<transcript>`` line per voice.
    Returns the path of the written file.
    """
    prompt_path = wav_dir / PROMPT_TEXT_FILENAME
    lines = [f"{name}|{text}" for name, text in transcripts.items()]
    prompt_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return prompt_path


def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None,
                           *, emit=None, cancel=None, on_cancel=None,
                           stall_timeout: Optional[float] = None) -> int:
    """Run a subprocess, streaming output to the console or to EMIT.

    With EMIT None the child inherits the real terminal and its output
    appears normally (the non-interactive CLI paths). With EMIT given (a
    ``callable(str)``) the child's stdout/stderr are merged, read line by
    line (splitting on both ``\\n`` and ``\\r`` so carriage-return progress
    updates like git's or tqdm's surface as lines), and each line is passed
    to EMIT — the in-TUI task view path.

    CANCEL is an optional ``threading.Event``: once set, ON_CANCEL (if given)
    is called (e.g. to touch a ``--cancel-file``), then the child's process
    group is terminated (SIGTERM, escalating to SIGKILL after a grace
    period) and 130 is returned.

    STALL_TIMEOUT (EMIT path only) is a no-output watchdog in seconds: when
    the child produces no new output line for that long, it is treated as
    wedged (a build whose compiler hung, a download that stopped moving) —
    the process group is terminated, an [ERROR] line is emitted, and 124 is
    returned so callers can report a stall distinctly from a plain failure.
    None (the default) waits forever, as before.

    Returns the process exit code.
    """
    import subprocess
    if emit is None:
        try:
            result = subprocess.run(argv,
                                    cwd=str(cwd) if cwd is not None else None)
        except OSError as exc:
            print(f"[ERROR] Could not run {' '.join(argv)}: {exc}")
            return 1
        return result.returncode

    popen_kwargs = {"stdout": subprocess.PIPE, "stderr": subprocess.STDOUT}
    if cwd is not None:
        popen_kwargs["cwd"] = str(cwd)
    if sys.platform == "win32":
        popen_kwargs["creationflags"] = \
            subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore[attr-defined]
    else:
        popen_kwargs["start_new_session"] = True
    try:
        proc = subprocess.Popen(argv, **popen_kwargs)
    except OSError as exc:
        emit(f"[ERROR] Could not run {' '.join(argv)}: {exc}")
        return 1

    cancelled = False
    stalled = False
    # Written by the reader thread, read by the poll loop below: a plain
    # float assignment is atomic enough under the GIL (no torn reads).
    last_output = time.monotonic()

    def _reader() -> None:
        nonlocal last_output
        try:
            pending = ""
            while True:
                # read1 (not read/readline) returns whatever a single pipe
                # read yields, without waiting to fill the buffer, and \r
                # is treated as a line break: carriage-return progress bars
                # (git clone --progress, tqdm/HuggingFace downloads) then
                # surface incrementally and keep the stall watchdog fed —
                # a readline-based reader blocked until the next \n would
                # let a healthy download starve to the timeout.
                chunk = proc.stdout.read1(4096)
                if not chunk:
                    break
                last_output = time.monotonic()
                pending += chunk.decode("utf-8", errors="replace")
                parts = re.split(r"[\r\n]+", pending)
                pending = parts.pop()
                for line in parts:
                    if line:
                        emit(line)
            if pending.strip():
                emit(pending)
        except (OSError, ValueError):
            pass

    reader = _spawn_reader(_reader)
    while True:
        if cancel is not None and cancel.is_set():
            cancelled = True
            if on_cancel is not None:
                try:
                    on_cancel()
                except Exception:
                    pass
                # Give a graceful-cancel hook (e.g. a --cancel-file) a
                # moment to let the child exit cleanly before forcing it.
                grace_end = time.time() + 3
                while time.time() < grace_end:
                    if proc.poll() is not None:
                        break
                    time.sleep(0.1)
            if proc.poll() is None:
                _terminate_process_group(proc)
            break
        if proc.poll() is not None:
            break
        if (stall_timeout is not None
                and time.monotonic() - last_output > stall_timeout):
            stalled = True
            emit(f"[ERROR] No output for {int(stall_timeout)}s — assuming "
                 "the process hung; stopping it.")
            _terminate_process_group(proc)
            break
        time.sleep(0.1)
    # The reader is a daemon: a join timeout here only means the child
    # closed its stdout but the thread is still draining — there is
    # nothing useful left to wait for.
    reader.join(timeout=5)
    if cancelled:
        return 130
    if stalled:
        return 124
    return proc.returncode


def _spawn_reader(target):
    import threading
    thread = threading.Thread(target=target, daemon=True)
    thread.start()
    return thread


def _terminate_process_group(proc) -> None:
    """Terminate PROC's process group (SIGTERM, then SIGKILL after a grace).

    Death is detected with ``proc.poll()`` (which reaps the zombie) rather
    than a ``killpg(pgid, 0)`` probe — the latter still succeeds on a
    zombie, so it would always wait the full grace period.
    """
    import signal
    if sys.platform == "win32":
        # TerminateProcess hits a single pid; children the server spawned
        # would survive as orphans. taskkill /T walks and kills the whole
        # process tree, then the poll loop reaps the direct child.
        try:
            subprocess.run(["taskkill", "/T", "/F", "/PID", str(proc.pid)],
                           capture_output=True, timeout=10)
        except (OSError, subprocess.SubprocessError):
            try:
                proc.terminate()
            except OSError:
                pass
        deadline = time.time() + 10
        while time.time() < deadline:
            if proc.poll() is not None:
                return
            time.sleep(0.1)
        try:
            proc.kill()
        except OSError:
            pass
        return
    try:
        pgid = os.getpgid(proc.pid)
    except (ProcessLookupError, OSError):
        return
    try:
        os.killpg(pgid, signal.SIGTERM)
    except (ProcessLookupError, OSError):
        return
    deadline = time.time() + 10
    while time.time() < deadline:
        if proc.poll() is not None:
            return
        time.sleep(0.1)
    try:
        os.killpg(pgid, signal.SIGKILL)
    except (ProcessLookupError, OSError):
        pass
    proc.wait()


def git_clone(url: str, target: Path, *, emit=None, cancel=None) -> int:
    """Clone URL into TARGET, streaming to the console or to EMIT. Returns
    the exit code."""
    if emit is None:
        print(f"[INFO] Cloning {url} into {target}...")
        return run_console_subprocess(["git", "clone", url, str(target)])
    emit(f"[INFO] Cloning {url} into {target}...")
    # --progress makes git report percentage updates even though stderr is
    # piped (it normally only does so on a terminal), feeding the task view.
    return run_console_subprocess(
        ["git", "clone", "--progress", url, str(target)],
        emit=emit, cancel=cancel)


def git_head(checkout: Path) -> Optional[str]:
    """CHECKOUT's current HEAD commit sha, or None when it is not a repo."""
    proc = run_console_subprocess_quiet(["git", "-C", str(checkout),
                                         "rev-parse", "HEAD"])
    if proc is None or proc.returncode != 0:
        return None
    return proc.stdout.decode("utf-8", errors="replace").strip() or None


def git_commit_time(checkout: Path) -> Optional[int]:
    """CHECKOUT's HEAD commit time as a unix timestamp, or None.

    Uses the *committer* time (``%ct``): a rebase or cherry-pick rewrites
    it to when the rewrite happened, so a force-pushed or rebased branch
    always looks newer than binaries built from the pre-rewrite sources.
    None (not a repo, probe failed) leaves the decision to the caller.
    """
    proc = run_console_subprocess_quiet(["git", "-C", str(checkout),
                                         "show", "-s", "--format=%ct",
                                         "HEAD"])
    if proc is None or proc.returncode != 0:
        return None
    try:
        return int(proc.stdout.decode("ascii", errors="replace").strip())
    except ValueError:
        return None


def git_update(checkout: Path, *, emit=None, cancel=None) -> int:
    """Update CHECKOUT to its remote's HEAD: fetch, then hard reset.

    The backend checkouts are read-only working copies of upstream repos —
    all state that matters (models, build trees, server.json, voices.json)
    is untracked and survives the reset, while local edits the installers
    made (the vendored-ggml patch in the audio.cpp checkout) are meant to
    be re-applied by the caller afterwards. ``git reset --hard`` is used
    instead of ``git pull`` because a pull merges against the working tree
    and would conflict on exactly those re-applied-by-design edits.

    The branch reset to is the remote's default (``refs/remotes/origin/
    HEAD``), falling back to ``main`` when the symbolic ref is missing (a
    bare-ish mirror or a restrictive server). EMIT/CANCEL behave like
    git_clone's (fetch runs with --progress so the task view sees updates).
    Returns the exit code of the first failing step (0 when the checkout
    now matches origin's HEAD).
    """
    if emit is None:
        print(f"[INFO] Updating git checkout {checkout}...")
    else:
        emit(f"[INFO] Updating git checkout {checkout}...")
    fetch_argv = ["git", "-C", str(checkout), "fetch"]
    reset_argv = ["git", "-C", str(checkout), "reset", "--hard"]
    if emit is not None:
        # --progress makes git report percentage updates even though stderr
        # is piped (it normally only does so on a terminal), feeding the
        # task view.
        fetch_argv.append("--progress")
    fetch_argv.append("origin")
    fetch_rc = run_console_subprocess(fetch_argv, emit=emit, cancel=cancel)
    if fetch_rc != 0:
        return fetch_rc
    branch = _origin_default_branch(checkout)
    return run_console_subprocess(reset_argv + [f"origin/{branch}"],
                                  emit=emit, cancel=cancel)


def _origin_default_branch(checkout: Path) -> str:
    """The remote's default branch name for CHECKOUT ("main" as fallback)."""
    proc = run_console_subprocess_quiet(
        ["git", "-C", str(checkout), "symbolic-ref",
         "refs/remotes/origin/HEAD"])
    if proc is not None and proc.returncode == 0:
        ref = proc.stdout.decode("utf-8", errors="replace").strip()
        # refs/remotes/origin/HEAD -> refs/remotes/origin/main
        name = ref.rpartition("/")[2]
        if name:
            return name
    return "main"


def run_console_subprocess_quiet(argv: List[str],
                                 cwd: Optional[Path] = None,
                                 timeout: Optional[float] = None):
    """Run ARGV silently and return the completed result.

    Unlike run_console_subprocess (which streams or returns only an exit
    code) this captures stdout and needs the process object itself, for the
    small git probes (rev-parse, symbolic-ref) whose *output* matters and
    whose failure is a normal, non-fatal outcome. TIMEOUT bounds the wait
    (e.g. for hardware probes like nvidia-smi that can hang on a wedged
    driver); a timeout kills the child and returns a failed result, not an
    exception. Returns None when the process could not be started.
    """
    import subprocess
    try:
        return subprocess.run(
            argv, capture_output=True,
            cwd=str(cwd) if cwd is not None else None, check=False,
            timeout=timeout)
    except subprocess.TimeoutExpired:
        class _TimedOut:
            returncode = -1
            stdout = b""
        return _TimedOut()
    except OSError:
        return None


def pip_install(packages: List[str], *, emit=None, cancel=None,
                env_dir: Optional[Path] = None,
                upgrade: bool = False) -> int:
    """pip install PACKAGES into a managed venv. Returns exit code.

    Delegates to ``backends.envs.pip_install`` so backend TTS packages are
    installed into their dedicated tool-managed environments (``envs/tts``
    default; ``envs/qwen`` / ``envs/faster`` via ENV_DIR) rather than into
    whatever interpreter happens to be running the wizard — and never two
    conflicting stacks into the same env. With UPGRADE pip runs with
    ``-U`` (the backend update action's freshness check: pip only installs
    when a newer version resolves, else reports "already satisfied").
    With EMIT given (the in-TUI task view) pip runs with its output streamed
    into EMIT; CANCEL aborts it. The import is local to avoid a circular
    import (envs imports this module).
    """
    from backends import envs
    return envs.pip_install(packages, emit=emit, cancel=cancel,
                            env_dir=env_dir, upgrade=upgrade)


def pip_uninstall(packages: List[str], *, emit=None,
                  env_dir: Optional[Path] = None) -> int:
    """pip uninstall PACKAGES from a managed venv. Returns exit code.

    Delegates to ``backends.envs.pip_uninstall`` (local import to avoid a
    circular import). Used by the backends' ``uninstall`` action. With EMIT
    given (the in-TUI task view) pip runs piped, streaming into EMIT, so
    its output never touches the terminal behind curses.
    """
    from backends import envs
    return envs.pip_uninstall(packages, emit=emit, env_dir=env_dir)