vlib/cmds_misc.py

"""Misc commands: list, info, remove, cleanup, auth, path, doctor."""

from pathlib import Path
import getpass
import json
import os
import sys
import textwrap
import time
import urllib.error
import urllib.request
from vlib import ctx
from vlib.ui import _confirm, _fail, _noninteractive, _note, _ok, _say, _step, _term_width, _trunc, _warn
from vlib.net import _get_hf_token, _hf_auth_headers
from vlib.tensors import read_safetensors
from vlib.sources import _base_paths
from vlib.registry import _iter_voices, _read_meta, _registry_resolve, log_operation


def cmd_list(args):
    if getattr(args, "target", None) == "base" or getattr(args, "list_base", False):
        _list_bases()
        return

    voices = [d for d in _iter_voices() if (d / "voice.json").exists()]
    if not voices:
        _say("  No voices stored yet. Use 'voice get <model>' to download one, or 'voice add <file> <json>' for local files.")
        return

    tw = _term_width()
    w_name = 26 if tw >= 80 else max(14, tw - 54)
    w_src = 40 if tw >= 80 else max(18, tw - w_name - 16)
    w_type = 8
    _say(f"  {_trunc('Name', w_name):<{w_name}} {_trunc('Source', w_src):<{w_src}} {_trunc('Type', w_type):<{w_type}}")
    _say(f"  {'─'*w_name} {'─'*w_src} {'─'*w_type}")
    for d in voices:
        meta = _read_meta(d / "voice.json") or {}
        name = meta.get("name", d.name)
        src = meta.get("source_hf_model") or meta.get("source") or "—"
        typ = meta.get("type", "voice")
        _say(f"  {_trunc(name, w_name):<{w_name}} {_trunc(src, w_src):<{w_src}} {_trunc(typ, w_type):<{w_type}}")


def _list_bases():
    base_root = ctx.VOICES_DIR / "base"
    if not base_root.exists() or not any(base_root.rglob("*.json")):
        _say("  No bases cached yet. Use 'voice delta <voice>' to create one.")
        return
    tw = _term_width()
    w_base = 40 if tw >= 80 else max(20, tw - 30)
    _say(f"  {_trunc('Base', w_base):<{w_base}} {'Bytes':<18} {'Fetched':<12}")
    _say(f"  {'─'*w_base} {'─'*18} {'─'*12}")
    seen = set()
    for jpath in sorted(base_root.rglob("*.json")):
        try:
            meta = json.loads(jpath.read_text())
        except Exception:
            continue
        parts = jpath.relative_to(base_root).parts
        base_id = f"{parts[0]}/{Path(parts[1]).stem}" if len(parts) == 2 else f"{parts[0]}/{parts[1]}"
        if base_id in seen:
            continue
        seen.add(base_id)
        b = meta.get("bytes")
        b_str = f"{b} ({b/1e6:.1f} MB)" if isinstance(b, int) else "—"
        fetched = meta.get("downloaded_at", "—")[:10]
        _say(f"  {_trunc(base_id, w_base):<{w_base}} {_trunc(b_str, 18):<18} {_trunc(fetched, 12):<12}")


def _dec_str(v):
    """Display coercion: GGUF-derived metadata sometimes holds raw byte
    lists (e.g. [108, 108, 97, 109, 97]); render them as text, never dumps."""
    if v is None:
        return None
    if isinstance(v, (bytes, bytearray)):
        try:
            return bytes(v).decode("ascii", "replace")
        except Exception:
            return str(v)
    if isinstance(v, list) and v and all(isinstance(x, int) for x in v):
        try:
            return bytes(v).decode("ascii", "replace")
        except Exception:
            return str(v)
    return v if isinstance(v, str) else str(v)

def _voice_tensors(meta):
    """Normalize voice metadata to a (tensors, synthetic) pair.
    Flat-only files (tensor_name/shape/dtype, no tensors list) become a
    synthetic 1-entry list so readers share one code path. Bare-string
    entries (hand-written json) are coerced so readers never traceback."""
    claims = meta.get("tensors")
    if isinstance(claims, list) and claims:
        fixed = [({"name": e} if isinstance(e, str) else e) for e in claims]
        return [e for e in fixed if isinstance(e, dict)], False
    if meta.get("tensor_name"):
        return [{"name": meta.get("tensor_name"), "shape": meta.get("shape"),
                 "dtype": meta.get("dtype")}], True
    return [], False


def _verify_voice_file(path, meta):
    """Cross-check voice.json claims against the artifact's own header.
    Returns warning strings (empty = agrees). Never raises: an unreadable
    file is a note, not a failure."""
    out = []
    if str(path).endswith(".gguf"):
        return out  # quant type names differ; nothing truthful to compare
    try:
        header, tens = read_safetensors(str(path))
    except Exception:
        return ["  ? Could not read the artifact header; showing voice.json as-is."]
    names = [k for k in tens if k != "delta.voice.marker"]
    claims, _synthetic = _voice_tensors(meta)
    if claims:
        for entry in claims:
            tname = entry.get("name")
            if tname not in tens:
                out.append(f"  ! MISMATCH tensor: voice.json lists '{tname}', file doesn't hold it.")
                continue
            info = tens[tname]
            claimed_shape = entry.get("shape")
            if claimed_shape is not None and list(claimed_shape) != list(info["shape"]):
                out.append(f"  ! MISMATCH shape '{tname}': voice.json says {list(claimed_shape)}, "
                           f"file holds {list(info['shape'])} — file wins.")
            claimed_dtype = (entry.get("dtype") or "").upper()
            if claimed_dtype and claimed_dtype != str(info["dtype"]).upper():
                out.append(f"  ! MISMATCH dtype '{tname}': voice.json says {entry.get('dtype')}, "
                           f"file holds {info['dtype']} — file wins.")
        claimed_names = {e.get("name") for e in claims}
        for extra in names:
            if extra not in claimed_names:
                out.append(f"  ! File holds '{extra}' which voice.json doesn't list.")
        if not names:
            out.append("  ? No tensors found in the artifact.")
        has_marker = "delta.voice.marker" in tens
        if meta.get("type") == "delta" and not has_marker:
            out.append("  ! voice.json claims delta but the file has no delta marker.")
        if meta.get("type") != "delta" and has_marker:
            out.append("  ! File carries a delta marker but voice.json type is "
                       f"'{meta.get('type', 'voice')}'.")
        return out
    # No tensor claims at all (minimal legacy json): only the delta-marker
    # check below is truthful; shape/dtype have nothing to compare against.
    if not names:
        return ["  ? No tensors found in the artifact."]
    has_marker = "delta.voice.marker" in tens
    if meta.get("type") == "delta" and not has_marker:
        out.append("  ! voice.json claims delta but the file has no delta marker.")
    if meta.get("type") != "delta" and has_marker:
        out.append("  ! File carries a delta marker but voice.json type is "
                   f"'{meta.get('type', 'voice')}'.")
    return out


def cmd_info(args):
    voicename = args.voicename.strip()
    voice_path, voice_meta = _registry_resolve(voicename)
    if voice_path is None or not voice_path.exists():
        bps = _base_paths(voicename)
        if bps and bps[0].exists():
            meta = _read_meta(bps[1]) or {}
            _say(f"  Name:    {voicename}")
            _say(f"  Type:    base")
            _say(f"  Source:  https://huggingface.co/{meta.get('source_hf_model', voicename)}")
            _say(f"  Tensor:  {meta.get('tensor_name', '—')}")
            _say(f"  Shape:   {meta.get('shape', '—')}")
            _say(f"  Dtype:   {meta.get('dtype', '—')}")
            _say(f"  File:    {bps[0]}")
            return
        _fail(f"  ✗ Voice '{voicename}' not found. Try: voice list")

    meta = _read_meta(Path(voice_path).parent / "voice.json") or voice_meta or {}
    name = meta.get("name", voicename)
    source = meta.get("source_hf_model") or meta.get("source") or "—"
    typ = meta.get("type", "voice")
    _say(f"  Name:    {name}")
    if meta.get("family") or meta.get("architecture"):
        _arch = _dec_str(meta.get("architecture"))
        _fam = _dec_str(meta.get("family"))
        _say(f"  Family:  {_fam or '—'}" + (f" ({_arch})" if _arch else ""))
    tensors, _synthetic = _voice_tensors(meta)
    only = tensors[0] if len(tensors) == 1 else {}
    if len(tensors) != 1:
        for entry in tensors:
            _say(f"    - {entry.get('name')}: {entry.get('shape')} · {entry.get('dtype')}")
    else:
        _say(f"  Tensor:  {meta.get('tensor_name') or only.get('name', '—')}")
        _say(f"  Shape:   {meta.get('shape') or only.get('shape', '—')}")
        _say(f"  Dtype:   {meta.get('dtype') or only.get('dtype', '—')}")
    b = meta.get("bytes")
    if isinstance(b, int):
        _say(f"  Bytes:   {b} ({b/1e6:.1f} MB)")
    if meta.get("method"):
        _say(f"  Method:  {meta.get('method')}")
        if meta.get("pulse"):
            _say(f"  Pulse:   {meta.get('pulse')}")
        if meta.get("donor"):
            _say(f"  Donor:   {meta.get('donor')}")
        chain = meta.get("chain")
        if isinstance(chain, list) and chain:
            _say(f"  Chain:   {' → '.join(str(c) for c in chain)}")
        cfg = meta.get("config")
        if isinstance(cfg, dict) and cfg:
            knobs = " · ".join(f"{k} {cfg[k]}" for k in
                               ("density", "epsilon", "lambda", "rescale", "seed")
                               if k in cfg)
            extra = " ".join(f"{k} {v}" for k, v in cfg.items()
                             if k not in ("density", "epsilon", "lambda", "rescale", "seed"))
            recipe = " · ".join(p for p in (knobs, extra) if p)
            chash = meta.get("config_hash")
            _say(f"  Config:  {recipe}" + (f" (#{chash})" if chash else ""))
    if meta.get("base"):
        _say(f"  Base:    {meta.get('base')}")
    if meta.get("license"):
        _say(f"  License: {_trunc(meta['license'], 80)}")
    _say(f"  Saved:   {meta.get('downloaded_at', meta.get('fetched_at', '—'))}")
    _say(f"  File:    {voice_path}")
    for w in _verify_voice_file(str(voice_path), meta):
        _warn(w)


def cmd_remove(args):
    name = args.voice.strip().lower()
    target = next((d for d in _iter_voices() if d.name.lower() == name), None)
    if target is None:
        _fail(f"  ✗ Voice '{args.voice}' not found. Try: voice list")
    if _noninteractive(args) and not getattr(args, "yes", False):
        _fail(f"  ✗ Can't remove '{target.name}' in non-interactive mode without --yes.")
    if not _confirm(f"  Remove voice '{target.name}'? [y/N]: ", args):
        _say("  Cancelled.")
        sys.exit(1)
    import shutil
    shutil.rmtree(target)
    log_operation({"op": "remove", "voice": target.name, "output": str(target), "status": "ok"})
    _ok(f"Removed '{target.name}'.")


def cmd_cleanup(args):
    parts_dir = ctx.VOICES_DIR / ".parts"
    if not parts_dir.exists() or not any(parts_dir.iterdir()):
        _say("No incomplete downloads to clean.")
        return
    removed, reclaimed = 0, 0
    import shutil
    force = bool(getattr(args, "force", False))
    for p in sorted(parts_dir.iterdir()):
        try:
            if p.is_dir():
                lock = p / ".lock"
                if lock.exists() and not force:
                    try:
                        if time.time() - lock.stat().st_mtime < 600:
                            _note(f"Skipping {p.name}: download in progress or stalled — use --force to clear")
                            continue
                    except Exception:
                        pass
                sz = sum(f.stat().st_size for f in p.rglob("*") if f.is_file())
                shutil.rmtree(p)
                removed += 1
                reclaimed += sz
                _ok(f"Removed {p.name}/ ({sz/1e6:.1f} MB)")
            else:
                sz = p.stat().st_size if p.is_file() else 0
                p.unlink()
                removed += 1
                reclaimed += sz
                _ok(f"Removed {p.name}")
        except Exception as e:
            if ctx.VERBOSE:
                _warn(f"Could not remove {p}: {e}")
    _ok(f"Cleaned {removed} entry(s), reclaimed {reclaimed/1e6:.1f} MB.")
    try:
        if not any(parts_dir.iterdir()):
            parts_dir.rmdir()
    except Exception:
        pass
def cmd_auth(args):
    tok_path = ctx.HF_TOKEN_PATH
    if getattr(args, "clear", False):
        if not tok_path.exists():
            _say("  No HF token stored — nothing to clear.")
            return
        try:
            tok_path.unlink(missing_ok=True)
            _ok("Removed stored HF token.")
        except Exception as e:
            _fail(f"  ✗ Could not remove token: {e}")
        return
    if getattr(args, "show", False):
        tok = _get_hf_token()
        if not tok or not tok.startswith("hf_") or len(tok) < 10:
            _say("  No valid HF token stored. Run: voice auth")
            return
        masked = tok[:4] + "·" * (len(tok) - 8) + tok[-4:] if len(tok) > 8 else "·" * len(tok)
        src = "environment" if os.environ.get("HF_TOKEN") or os.environ.get("HF_HUB_TOKEN") else str(tok_path)
        _say(f"  ✓ HF token present: {masked} ({src})")
        return
    if not sys.stdin.isatty():
        _fail("  ✗ voice auth needs an interactive terminal (stdin is piped).")
    _say("  Paste your HF token (hf_…): it stays on this machine, 0600, never logged.")
    try:
        try:
            tok = getpass.getpass("  Paste HF token: ").strip()
        except Exception:
            tok = input("  Paste HF token: ").strip()
    except (EOFError, KeyboardInterrupt):
        _say("  ⚠ Cancelled.")
        sys.exit(1)
    if not tok:
        _say("  ⚠ No token entered.")
        sys.exit(1)
    if not tok.startswith("hf_") or len(tok) < 10:
        _fail("  ✗ That doesn't look like an HF token (should start with hf_…).")
    _step("Verifying…")
    try:
        req = urllib.request.Request("https://huggingface.co/api/whoami-v2",
                                     headers=_hf_auth_headers("https://huggingface.co/api/whoami-v2", {"Authorization": f"Bearer {tok}"}))
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read())
            _ok(f"Verified as {data.get('name') or data.get('id') or 'ok'}.")
    except urllib.error.HTTPError as e:
        if e.code in (401, 403):
            _fail("  ✗ Token rejected (401). Check it at https://huggingface.co/settings/tokens")
        _warn(f"Verification skipped (HTTP {e.code}), saving anyway.")
    except Exception:
        _warn("Could not verify now — saving anyway.")
    try:
        ctx.VOICES_DIR.mkdir(parents=True, exist_ok=True)
        try:
            os.chmod(ctx.VOICES_DIR, 0o700)
        except Exception:
            pass
        tmp = tok_path.with_suffix(".tmp")
        # atomic 0600: create with restricted perms, avoid 0644 race
        fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            os.write(fd, (tok.strip() + "\n").encode("utf-8"))
            try:
                os.fsync(fd)
            except Exception:
                pass
        finally:
            os.close(fd)
        try:
            os.chmod(tmp, 0o600)
        except Exception:
            pass
        tmp.replace(tok_path)
        try:
            os.chmod(tok_path, 0o600)
        except Exception:
            pass
        _ok(f"Saved to {tok_path} (0600).")
    except Exception as e:
        _fail(f"  ✗ Could not save token: {e}")


def cmd_path(_args):
    script = _entry_script()
    if script is None:
        _fail("  ✗ Cannot locate the voice entry script in this interpreter. Run voice.py directly.")
    script = Path(script).resolve()
    vlib_src = script.parent / "vlib"
    if sys.platform.startswith("win"):
        import shutil as _sh
        bindir = Path(os.environ.get("LOCALAPPDATA", "")) / "voice-cli"
        if str(bindir) == "voice-cli":
            _fail("  ✗ Couldn't find a Windows app-data folder to install into.")
        bindir.mkdir(parents=True, exist_ok=True)
        _copy_tree(vlib_src, bindir / "vlib")
        _sh.copy2(str(script), str(bindir / "voice.py"))
        if _sh.which("voice"):
            _ok("Voice updated on your PATH.")
        else:
            _ok("Voice is ready. Add this to your PATH (PowerShell):")
            _say(f'    setx PATH "%PATH%;{bindir}"')
        return
    import shutil
    existing = shutil.which("voice")
    if existing:
        try:
            same = Path(existing).resolve() == script
        except Exception:
            same = False
        if same:
            _ok("Voice is already on your PATH.")
            return
        # Stale install: relink only when the old entry is our symlink
        # (points at a voice.py); never touch a foreign binary.
        try:
            old_target = os.readlink(existing)
        except OSError:
            _fail(f"  ✗ 'voice' on PATH isn't ours ({existing}). Remove or rename it, then retry.")
        if Path(old_target).name != "voice.py":
            _fail(f"  ✗ '{existing}' isn't a voice install (points at {old_target}). Remove it (e.g. sudo mv {existing} {existing}.bak), then retry.")
        try:
            Path(existing).unlink()
            Path(existing).symlink_to(str(script))
        except (PermissionError, OSError):
            _fail(f"  ✗ Couldn't relink {existing}. Try: sudo ln -sf {script} {existing}")
        _ok(f"Voice updated: {existing} now points at {script}. Open a new terminal and run: voice")
        return
    candidates = [Path("/usr/local/bin"), Path.home() / ".local" / "bin", Path("/usr/bin"), Path("/bin")]
    on_path = next((d for d in candidates if d.is_dir() and str(d) in os.environ.get("PATH", "").split(":")), None)
    if on_path is None:
        _fail("  ✗ No writable directory on PATH found. Add one and retry.")
    target = on_path / "voice"
    try:
        target.symlink_to(str(script))
    except FileExistsError:
        target.unlink()
        target.symlink_to(str(script))
    except (PermissionError, OSError):
        _fail(f"  ✗ Couldn't create a link in {on_path}. Try: sudo ln -s {script} {target}")
    _ok(f"Voice is ready at {target}. Open a new terminal and run: voice")


def cmd_doctor(_args):
    import platform
    import shutil
    _say(f"  voice {ctx.VERSION} — doctor")
    _say(f"  Python {platform.python_version()} ({sys.executable})")
    uv = shutil.which("uv")
    _say(f"  uv {'found: ' + uv if uv else 'NOT FOUND — curl -LsSf https://astral.sh/uv/install.sh | sh'}")
    venv = ctx.VOICES_DIR / ".venv"
    if (venv / "bin" / "python").exists():
        _say(f"  venv {venv} present")
    else:
        _say(f"  venv not created yet (created on first get/cast/delta)")
    try:
        free = shutil.disk_usage(ctx.VOICES_DIR if ctx.VOICES_DIR.exists() else Path.home()).free
        _say(f"  disk free {free/1e9:.1f} GB")
    except Exception:
        pass
    tok = _get_hf_token()
    _say(f"  HF token {'present' if tok else 'not present'}")
    if ctx.OPERATIONS_LOG.exists():
        try:
            n = sum(1 for _ in open(ctx.OPERATIONS_LOG))
            _say(f"  operations logged: {n} ({ctx.OPERATIONS_LOG})")
        except Exception:
            pass
    if ctx.REPORTS_JSONL.exists():
        try:
            n = sum(1 for _ in open(ctx.REPORTS_JSONL))
            pending = len(list(ctx.REPORTS_OUTBOX.glob("*.json"))) if ctx.REPORTS_OUTBOX.exists() else 0
            _say(f"  reports: {n} local, {pending} pending in outbox ({ctx.REPORTS_JSONL})")
            if pending > 5000:
                _warn(f"  outbox >5000 pending — run: voice report --sync --yes")
            # headroom
            try:
                sz = ctx.REPORTS_JSONL.stat().st_size
                if sz > 50 * 1024 * 1024:
                    _warn(f"  reports.jsonl {sz/1e6:.1f} MB — consider archiving")
            except Exception:
                pass
        except Exception:
            pass
    else:
        _say(f"  reports: none yet — run: voice report")
    _step("Checking connectivity…")
    try:
        req = urllib.request.Request("https://huggingface.co/api/whoami-v2" if tok else "https://huggingface.co",
                                     headers=_hf_auth_headers("https://huggingface.co"))
        with urllib.request.urlopen(req, timeout=3) as resp:
            _say(f"  connectivity huggingface.co ok ({resp.status})")
    except Exception as e:
        _warn(f"connectivity failed: {e}")
    # check push target
    try:
        _say(f"  community reports: {ctx.REPORTS_DATASET_ID} (share prompt on voice report)")
    except Exception:
        pass


def _print_version():
    import platform
    _say(f"  voice {ctx.VERSION}")
    _say(f"  Python {platform.python_version()} ({sys.executable})")
    for mod in ("gguf", "numpy", "huggingface_hub", "hf_xet"):
        try:
            m = __import__(mod)
            _say(f"  {mod} {getattr(m, '__version__', 'installed')}")
        except Exception:
            _say(f"  {mod} not installed")


def _print_completion(shell):
    cmds = ["get", "cast", "delta", "graft", "chorus", "add", "list", "info", "remove", "cleanup", "auth", "report", "path", "doctor"]
    voices = [d.name for d in _iter_voices() if (d / "voice.json").exists()]
    vlist = " ".join(voices)
    clist = " ".join(cmds)
    if shell == "bash":
        print(textwrap.dedent(f"""\
            _voice_completions() {{
                local cur="${{COMP_WORDS[COMP_CWORD]}}"
                local cmds="{clist}"
                if [[ $COMP_CWORD -eq 1 ]]; then
                    COMPREPLY=($(compgen -W "$cmds --help --version --completion" -- "$cur"))
                    return
                fi
                case "${{COMP_WORDS[1]}}" in
                    cast|delta|graft|chorus|info|remove) COMPREPLY=($(compgen -W "{vlist}" -- "$cur")) ;;
                    list) COMPREPLY=($(compgen -W "base --base" -- "$cur")) ;;
                    *) COMPREPLY=() ;;
                esac
            }}
            complete -F _voice_completions voice
            """))
    elif shell == "zsh":
        print(textwrap.dedent(f"""\
            #compdef voice
            _voice() {{
                local -a cmds; cmds=({clist})
                local -a voices; voices=({vlist})
                _arguments -C "1: :($cmds)" "*::arg:->args"
                case $line[1] in
                    cast|delta|graft|chorus|info|remove) _describe voices voices ;;
                    list) _values "target" base ;;
                esac
            }}
            compdef _voice voice
            """))
    elif shell == "fish":
        print(textwrap.dedent(f"""\
            complete -c voice -f
            complete -c voice -n "__fish_use_subcommand" -a "{clist}" -d "voice command"
            complete -c voice -n "__fish_seen_subcommand_from cast delta graft chorus info remove" -a "{vlist}"
            """))


def _copy_tree(src, dst):
    """Refresh-copy a directory (Windows PATH install needs vlib/ beside voice.py)."""
    import shutil as _sh
    if Path(src).is_dir():
        _sh.copytree(str(src), str(dst), dirs_exist_ok=True,
                     ignore=_sh.ignore_patterns("__pycache__"))
    return dst


def _entry_script():
    """Absolute path of the CLI entry (voice.py), or None if unknown.
    The split moved commands into vlib/, so __file__ here is the module —
    resolve via __main__ instead. Single seam for every entry-path need."""
    _main = getattr(sys.modules.get("__main__"), "__file__", None)
    if not _main:
        return None
    return os.path.realpath(_main)


def _ensure_venv(cmd):
    if os.environ.get("VOICE_NO_VENV") in ("1", "true", "True"):
        return
    if cmd not in ("get", "cast", "delta"):
        return
    venv = ctx.VOICES_DIR / ".venv"
    venv_py = venv / "bin" / "python"
    try:
        if venv_py.exists() and (sys.prefix == str(venv) or Path(sys.executable).resolve().parent.parent == venv.resolve()):
            return
    except Exception:
        pass
    import shutil
    if shutil.which("uv") is None:
        _fail("  ✗ uv not found. Install it, then retry:\n    curl -LsSf https://astral.sh/uv/install.sh | sh")
    if venv_py.exists():
        try:
            _main = _entry_script()
            if not _main:
                _warn(" venv re-exec skipped: entry script unknown, continuing in-process.")
                return
            os.execv(str(venv_py), [str(venv_py), _main, *sys.argv[1:]])
        except Exception as e:
            if ctx.VERBOSE:
                _warn(f"could not re-exec into venv: {e}")
            return
    _step(f"Creating a virtual environment at {venv}…")
    import subprocess
    try:
        subprocess.run(["uv", "venv", str(venv), "--python", sys.executable], capture_output=True, check=True)
        subprocess.run(["uv", "pip", "install", "--quiet", "--python", str(venv_py),
                        "gguf", "numpy", "huggingface_hub[hf_xet]", "hf_xet"], capture_output=True, check=True)
    except Exception as e:
        _fail(f"  ✗ Failed to create venv: {e}")
    _main = _entry_script()
    if not _main:
        _warn(" venv re-exec skipped: entry script unknown, continuing in-process.")
        return
    try:
        os.execv(str(venv_py), [str(venv_py), _main, *sys.argv[1:]])
    except Exception as e:
        _fail(f"  ✗ Could not re-exec into venv: {e}")