vlib/compute.py

"""Cast math: orient, fuse, quantize, pack writes."""

from pathlib import Path
import os
import struct
import sys
import time
from vlib import ctx
from vlib.ui import _confirm, _fail, _noninteractive, _note, _ok, _progress_pref, _say, _step, _transform, _warn
from vlib.tensors import _gguf_raw_to_f32, _is_delta_safetensors, _is_ggml_quant, _is_gguf_file, _safetensors_write_repl, decode_to_f32, encode_from_f32, read_safetensors, rewrite_safetensors_tensor
from vlib.sources import OUTPUT_ALIASES, _nested, detect_output_tensor, largest_2d, open_source
from vlib.registry import log_operation


def _load_voice(voice_path):
    """Load a voice artifact. Returns (f32_array_or_None, ref, raw_bytes_or_None, ggml_type_or_None).
    Quantized GGUF artifacts carry raw blocks; safetensors artifacts decode to f32."""
    src = open_source(str(voice_path))
    names = [n for n in src.names() if n != "delta.voice.marker"]
    if not names:
        _fail(f"  ✗ No tensors found in {voice_path}.")
    tname = names[0] if len(names) == 1 else "output.weight" if "output.weight" in names else names[0]
    ref = src.ref(tname)
    if src.kind == "gguf" and _is_ggml_quant(ref.dtype):
        return None, ref, src.read_raw(tname), ref.dtype
    return src.read_f32(tname), ref, None, None


def _orient(arr, shape, target_shape):
    import numpy as np
    if tuple(shape) == tuple(target_shape):
        return arr
    if tuple(shape) == tuple(target_shape[::-1]):
        if ctx.VERBOSE:
            _note("Transposing to match the target layout…")
        return np.ascontiguousarray(arr.T)
    return None


def _cast_st_prepare(target, tensors, vf32, vref, meta, args, vraw=None, vqtype=None, strict_target=False, target_name=None):
    """Per-tensor compute for a safetensors target. Returns (tname, data, tdtype, tshape).
    target_name (from _resolve_pack_targets) overrides lookup; None keeps the
    original single-tensor behavior byte-for-byte."""
    import numpy as np
    if vraw is not None:
        # quant artifact -> safetensors target: dequant is the only requant path
        from gguf.constants import GGMLQuantizationType as Q
        from gguf.quants import quant_shape_to_byte_shape
        vf32 = _gguf_raw_to_f32(vraw, Q[vqtype], tuple(reversed(tuple(int(x) for x in vref.shape))))
    names = list(tensors.keys())
    if target_name is not None:
        tname = target_name if target_name in names else None
        if tname is None:
            _fail(f"  ✗ Resolved target '{target_name}' for '{vref.name}' not in {target}.")
    else:
        tname = vref.name if vref.name in names else None
        if tname is None and strict_target:
            _fail(f"  ✗ Tensor '{vref.name}' not found in target {target}. Voicepack requires exact match (not mapping to another tensor).")
    if tname is None:
        tname = detect_output_tensor(names, None) or largest_2d(names, {n: tensors[n]["shape"] for n in names})
    if tname is None:
        _fail("  ✗ Could not find a matching tensor in the target.")
    tshape = tuple(tensors[tname]["shape"])
    tdtype = tensors[tname]["dtype"]
    # --quant override for safetensors target: only F32/F16/BF16/keep/target make sense (no GGUF quants).
    _qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
    if _qo is None:
        _fail(f"  ✗ Invalid --quant '{getattr(args, 'quant', None)}' for safetensors target. Use keep|target|f32|f16|bf16.")
    if _qo not in ("keep", "target", "f32", "f16", "bf16"):
        _fail(f"  ✗ --quant {_qo} needs GGUF target (safetensors has no quant {_qo}). Use keep|target|f32|f16|bf16.")
    if _qo in ("f32", "f16", "bf16"):
        _want = {"f32": "F32", "f16": "F16", "bf16": "BF16"}[_qo]
        if _want != tdtype.upper():
            from vlib.tensors import dtype_bytes as _db
            try:
                _same_size = _db(_want) == _db(tdtype.upper())
            except Exception:
                _same_size = False
            if not _same_size:
                _fail(f"  ✗ --quant {_qo} would change '{tname}' {tdtype} → {_want} (different byte size; safetensors casts keep byte size). Use --quant keep|target.")
            _warn(f"  Re-encoding target {tdtype} → {_want} for --quant {_qo} (extra conversion).")
            tdtype = _want
    # Show quant/target by default for tensor commands (advanced) and verbose for voice.
    try:
        _svq = vqtype if vraw is not None else vref.dtype
    except Exception:
        _svq = "?"
    if ctx.VERBOSE:
        _note(f"Voice '{vref.name}' is {_svq} {tuple(vref.shape)} → target '{tname}' is {tdtype} {tshape}")
    if vraw is not None and ctx.VERBOSE:
        _note(f"Dequanting {vqtype} once to F32 for safetensors target (unavoidable, safetensors has no quant).")
    oriented = _orient(vf32, vref.shape, tshape)
    if oriented is None:
        _fail(f"  ✗ Voice shape {vref.shape} doesn't fit target tensor {tname} {tshape}.")
    if (meta or {}).get("type") == "delta":
        _step(f"Fusing style into {tname} (delta + head → {tdtype}, kept)…")
        with open(target, "rb") as f:
            _hs = struct.unpack("<Q", f.read(8))[0]
        info = tensors[tname]
        with open(target, "rb") as f:
            f.seek(8 + _hs + info["data_offsets"][0])
            head_f32 = decode_to_f32(f.read(info["data_offsets"][1] - info["data_offsets"][0]), tdtype, tshape)
        oriented = np.ascontiguousarray(head_f32 + oriented)
    data = encode_from_f32(oriented, tdtype)
    return tname, data, tdtype, tshape


def _cast_core(target, target_format, out_path, vf32, vref, meta, args, op="cast", vraw=None, vqtype=None, strict_target=False, target_name=None):
    if getattr(args, "speak", False) and target_format != "gguf":
        _fail("  ✗ --speak needs a .gguf target. Re-run without --speak for .safetensors, or point cast at a .gguf file.")
    if target_format == "gguf":
        _cast_gguf(target, out_path, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype, strict_target=strict_target, target_name=target_name)
        return
    # safetensors target
    header, tensors = read_safetensors(target)
    t0s = time.time()
    tname, data, tdtype, tshape = _cast_st_prepare(target, tensors, vf32, vref, meta, args,
                                                  vraw=vraw, vqtype=vqtype, strict_target=strict_target,
                                                  target_name=target_name)
    _step(f"Replacing {tname} ({tdtype} {tshape})…")
    rewrite_safetensors_tensor(target, out_path, tname, data, tdtype, tshape)
    file_size = os.path.getsize(out_path)
    try:
        _mb = len(data) / 1e6
        _rate = f" @ {(_mb) / max(time.time() - t0s, 1e-6):.0f} MB/s" if _mb >= 0.1 else ""
        _note(f"Kept '{tname}': {_mb:.1f} MB → {tdtype}{_rate}")
    except Exception:
        pass
    log_operation({"op": op, "target": target, "format": "safetensors", "tensor": tname,
                   "dtype": tdtype, "shape": list(tshape), "output": out_path, "status": "ok"})
    _ok(f"Done → {out_path}")
    _say(f"  Saved → {out_path} ({tdtype} {list(tshape)})")


def _is_head_name(name):
    if not name:
        return False
    try:
        if name in OUTPUT_ALIASES or name in ("token_embd.weight",):
            return True
        for a in OUTPUT_ALIASES:
            if name.endswith(a):
                return True
    except Exception:
        pass
    return False


def _parse_quant_override(s):
    # Returns normalized key: keep|target or Q-type name upper (e.g. Q4_0). Trivial: accept all GGML quants.
    if s is None:
        return "keep"
    k = str(s).strip().lower().replace("-", "_")
    if k in ("keep", "target"):
        return k
    # Aliases for float family.
    if k in ("fp32",):
        k = "f32"
    if k in ("fp16",):
        k = "f16"
    # Try any GGMLQuantizationType name.
    try:
        from gguf.constants import GGMLQuantizationType as _Qp
        # Case-insensitive match against all Q names.
        for _q in _Qp:
            if _q.name.lower() == k:
                return k
    except Exception:
        pass
    # Fallback known list (when gguf import fails).
    _known = {"f32", "f16", "bf16", "q8_0", "q8_1", "q8_k", "q1_0", "q4_0", "q4_1", "q5_0", "q5_1",
              "q2_k", "q3_k", "q4_k", "q5_k", "q6_k", "iq1_s", "iq1_m", "iq2_xxs", "iq2_xs", "iq2_s",
              "iq3_xxs", "iq3_s", "iq4_nl", "iq4_xs", "tq1_0", "tq2_0", "mxfp4", "nvfp4"}
    if k in _known:
        return k
    return None


def _quant_from_key(key):
    # key normalized via _parse_quant_override (lower). Returns Q type or None for keep/target.
    if key in ("keep", "target", None):
        return None
    try:
        from gguf.constants import GGMLQuantizationType as _Qq
        for _q in _Qq:
            if _q.name.lower() == key:
                return _q
        # Upper direct (e.g. Q4_0).
        return _Qq[key.upper()]
    except Exception:
        return None


def _qerr(e):
    """Readable lib error: gguf often raises bare NotImplementedError ('')."""
    s = str(e).strip()
    return f"{type(e).__name__}: {s}" if s else type(e).__name__


_NEAREST_CACHE = {}


def _nearest_encodable(target_q):
    """Nearest gguf-encodable quant to target_q by bits/param (ties → larger =
    quality-preserving). The lib decodes everything but encodes only a subset
    (no K/IQ quants); this keeps fused tensors near the original size instead
    of ballooning to F16. Support probed once per process (version-proof)."""
    import numpy as np
    from gguf.constants import GGMLQuantizationType as Q
    from gguf.constants import GGML_QUANT_SIZES as S
    from gguf.quants import quantize
    key = getattr(target_q, "name", str(target_q))
    if key in _NEAREST_CACHE:
        return _NEAREST_CACHE[key]
    tbpp = S[target_q][1] * 8 / S[target_q][0]
    probe = (np.arange(64 * 64, dtype=np.float32).reshape(64, 64) - 2048) / 64
    scored = []
    for name in ("Q8_0", "Q5_1", "Q5_0", "Q4_1", "Q4_0", "MXFP4", "F16"):
        cand = getattr(Q, name, None)
        if cand is None or cand == target_q:
            continue
        try:
            quantize(probe, cand)
        except Exception:
            continue
        cbpp = S[cand][1] * 8 / S[cand][0]
        scored.append((abs(cbpp - tbpp), -cbpp, cand))
    best = min(scored)[2] if scored else None
    _NEAREST_CACHE[key] = best
    return best

def _quantize_to_Q(fused, req_q):
    # Quantize F32 fused array to any Q (F32/F16/BF16 via astype, else gguf quantize).
    # Returns (q_head, out_qtype). Raises on unsupported (e.g. Q1_0/Q8_1/Q8_K NotImplemented).
    import numpy as np
    from gguf.constants import GGMLQuantizationType as Q
    from gguf.quants import quantize
    if req_q in (Q.F32,):
        return fused.astype(np.float32), Q.F32
    if req_q in (Q.F16,):
        return fused.astype(np.float16), Q.F16
    try:
        _b = getattr(Q, "BF16", None)
    except Exception:
        _b = None
    if _b is not None and req_q == _b:
        _u32 = fused.astype(np.float32).view(np.uint32)
        return (((_u32 + 0x7FFF + ((_u32 >> 16) & 1)) >> 16).astype(np.uint16), _b)
    # All other GGML quants (Q4_0/Q4_K/Q6_K/IQ*/TQ*/MXFP4/...) via single quantize call (trivial).
    return quantize(fused, req_q), req_q


def _fuse_delta(target_tensor, delta_f32, strict=False):
    """Dequant the target head once to F32 and fuse the delta (head + delta).
    Single seam for the three dequant→fuse sites in _gguf_fuse_one.
    strict (safetensors delta path only): fail clean on shape mismatch, exactly as before."""
    import numpy as np
    from gguf.constants import GGMLQuantizationType as Q
    if target_tensor.tensor_type in (Q.F32, Q.F16):
        try:
            head_f32 = target_tensor.data.astype(np.float32)
        except Exception:
            head_f32 = np.asarray(target_tensor.data, dtype=np.float32)
    else:
        from gguf.quants import dequantize as _deq
        head_f32 = _deq(target_tensor.data, target_tensor.tensor_type)
    if strict and head_f32.shape != delta_f32.shape:
        sys.stdout.write("\n")
        _fail(f"  ✗ Style shape {delta_f32.shape} doesn't match model head {head_f32.shape}.")
    return np.ascontiguousarray(head_f32 + delta_f32)


def _gguf_fuse_one(target, target_tensor, tname, logical, vf32, vref, meta, args, op, vraw=None, vqtype=None):
    """Fuse one voice tensor into its target GGUF tensor form. Returns (q_head, out_qtype)."""
    import numpy as np
    from gguf.constants import GGMLQuantizationType as Q
    from gguf.quants import quantize
    # Philosophy: preserve quality, avoid dequant->quant cycles.
    # - GGUF quant voice (vraw) is kept at its obtained quant (never upscale Q2->Q6, no inventing data).
    # - Safetensors high-fidelity lm_head (voice) is quantized once to Q8_0.
    # - Generic tensors keep target quant (F32 norms stay F32, never forced Q8_0).
    # - All GGML quants supported (trivial passthrough + quantize where implemented).
    _quant_override = _parse_quant_override(getattr(args, "quant", None) or "keep")
    if _quant_override is None:
        _fail(f"  ✗ Invalid --quant '{getattr(args, 'quant', None)}'. Use keep|target|<any quant like q8_0,q4_0,q4_k,q6_k,iq4_nl,f16,f32,bf16>.")
    _is_head = _is_head_name(vref.name) or _is_head_name(tname)
    _is_delta = (meta or {}).get("type") == "delta"
    # Show quant/target by default for advanced tensor commands (and verbose for voice)
    try:
        _tq = target_tensor.tensor_type.name
    except Exception:
        _tq = "?"
    try:
        _vq = vqtype if vraw is not None else vref.dtype
    except Exception:
        _vq = "?"
    if ctx.VERBOSE:
        _note(f"Voice '{vref.name}' is {_vq} {tuple(vref.shape)} → target '{tname}' is {_tq} {logical}")
    if vraw is not None:
        # GGUF quant voice: keep obtained quant (no upscale, no extra cycle).
        try:
            _vqtype = Q[vqtype]
        except Exception:
            _fail(f"  ✗ Unknown voice quant '{vqtype}'.")
        if tuple(vref.shape) != logical:
            _fail(f"  ✗ Voice shape {vref.shape} doesn't fit target tensor {tname} {logical}.")
        if _quant_override in ("keep",):
            # Transplant quant type as-is (may differ from target type — intentional, preserves quality).
            if _vqtype.name != target_tensor.tensor_type.name and ctx.VERBOSE:
                _note(f"Keeping voice quant {_vqtype.name} (target was {target_tensor.tensor_type.name}, not upscaling).")
            _step(f"Casting onto {os.path.basename(target)} ({_vqtype.name}, kept)…")
            from gguf.quants import quant_shape_to_byte_shape as _q2b_keep
            _keep_shape = _q2b_keep(tuple(int(x) for x in vref.shape), _vqtype)
            _need = 1
            for _d in _keep_shape:
                _need *= _d
            _raw = np.frombuffer(vraw, dtype=np.uint8)
            if _raw.size < _need:
                _fail(f"  ✗ Short voice data for transplant ({_raw.size}/{_need} bytes).")
            q_head, out_qtype = _raw[:_need].reshape(_keep_shape), _vqtype
        elif _quant_override == "target":
            # Explicit: dequant once then quant to target (extra cycle, warn).
            _warn(f"Re-quanting voice {_vqtype.name} → target {target_tensor.tensor_type.name} (extra dequant→quant cycle).")
            from gguf.quants import dequantize as _deq2
            from gguf.quants import quant_shape_to_byte_shape as _q2b
            # vref.shape is logical already checked
            _rshape = tuple(vref.shape)
            try:
                _f32v = _deq2(np.frombuffer(vraw, dtype=np.uint8).reshape(_q2b(_rshape, _vqtype)), _vqtype)
            except Exception as e:
                _fail(f"  ✗ Could not dequant voice for --quant target: {e}")
            if _is_delta:
                _f32v = _fuse_delta(target_tensor, _f32v, strict=True)
            t_quant0 = time.time()
            _tq2 = target_tensor.tensor_type
            if _tq2 in (Q.F32,):
                q_head, out_qtype = _f32v.astype(np.float32), Q.F32
            elif _tq2 in (Q.F16,):
                q_head, out_qtype = _f32v.astype(np.float16), Q.F16
            else:
                try:
                    _bf = getattr(Q, "BF16", None)
                except Exception:
                    _bf = None
                if _bf is not None and _tq2 == _bf:
                    # BF16 target: gguf lib has no BF16 encoder here; keep F32 to avoid loss
                    q_head, out_qtype = _f32v.astype(np.float32), Q.F32
                    _warn("  BF16 target: keeping F32 to avoid loss (use --quant f16/bf16 explicitly if needed).")
                else:
                    try:
                        q_head, out_qtype = quantize(_f32v, _tq2), _tq2
                    except Exception as e:
                        _fail(f"  ✗ Could not quant to target {_tq2.name}: {e}")
            _note(f"Re-quantized {_f32v.nbytes/1e6:.1f} MB → {_tq2.name} @ {(_f32v.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
        else:
            # Explicit q8_0/f16/f32/bf16 for quant voice: dequant once then quant.
            from gguf.quants import dequantize as _deq2
            from gguf.quants import quant_shape_to_byte_shape as _q2b
            _rshape = tuple(vref.shape)
            try:
                _f32v = _deq2(np.frombuffer(vraw, dtype=np.uint8).reshape(_q2b(_rshape, _vqtype)), _vqtype)
            except Exception as e:
                _fail(f"  ✗ Could not dequant voice for --quant {_quant_override}: {e}")
            if _is_delta:
                _f32v = _fuse_delta(target_tensor, _f32v, strict=True)
            _qn = _quant_from_key(_quant_override)
            if _qn is None:
                _fail(f"  ✗ Invalid --quant '{_quant_override}'.")
            t_quant0 = time.time()
            try:
                q_head, out_qtype = _quantize_to_Q(_f32v, _qn)
            except NotImplementedError as e:
                _fail(f"  ✗ Quant {getattr(_qn, 'name', _quant_override)} not implemented by gguf lib ({e}). Keeping original is recommended.")
            except Exception as e:
                _fail(f"  ✗ Could not quant to {_quant_override}: {e}")
            _note(f"Quantized {_f32v.nbytes/1e6:.1f} MB → {out_qtype.name} @ {(_f32v.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
    else:
        oriented = _orient(vf32, vref.shape, logical)
        if oriented is None:
            _fail(f"  ✗ Voice shape {vref.shape} doesn't fit target tensor {tname} {logical}.")
        t_fuse0 = time.time()
        # Decide out quant: Q8_0 once only for safetensors high-fidelity heads; else keep target.
        # vref.dtype here is F32/F16/BF16 (high-fidelity, since vraw is None).
        _target_q = target_tensor.tensor_type
        # Explicit override wins (any quant trivially supported via _quantize_to_Q).
        _req = None
        if _quant_override != "keep":
            if _quant_override == "target":
                _req = _target_q
            else:
                _req = _quant_from_key(_quant_override)
                if _req is None:
                    _fail(f"  ✗ Invalid --quant '{_quant_override}'.")
        t_fuse0 = time.time()
        if _is_delta:
            # Fuse first (F32 math preserves quality), decide quant after.
            fused = _fuse_delta(target_tensor, oriented, strict=True)
            if _req is not None:
                # Explicit --quant for delta (all quants trivial via helper).
                _step(f"Fusing style into {tname} (delta + head → {_req.name})…")
                t_quant0 = time.time()
                try:
                    q_head, out_qtype = _quantize_to_Q(fused, _req)
                except NotImplementedError as e:
                    _fail(f"  ✗ Quant {_req.name} not implemented by gguf lib ({e}). Keeping original is recommended.")
                except Exception as e:
                    _fail(f"  ✗ Could not quant delta to {_req.name}: {e}")
                _note(f"Quantized {fused.nbytes/1e6:.1f} MB → {_req.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
            elif _is_head:
                # Head delta from high-fidelity: Q8_0 once (old behavior, with %32 fallback).
                if fused.shape[-1] % 32 != 0:
                    _warn(f"Head last dim {fused.shape[-1]} not ×32 — Q8_0 needs %32, keeping F16 (larger, compatible).")
                    _step(f"Fusing style into {tname} (delta + head → F16 fallback)…")
                    q_head, out_qtype = fused.astype(np.float16), Q.F16
                else:
                    _step(f"Fusing style into {tname} (delta + head → Q8_0)…")
                    t_quant0 = time.time()
                    try:
                        q_head, out_qtype = quantize(fused, Q.Q8_0), Q.Q8_0
                    except Exception as e:
                        _fail(f"  ✗ Could not quant delta to Q8_0: {e}")
                    _note(f"Quantized {fused.nbytes/1e6:.1f} MB → Q8_0 @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
            else:
                # Generic delta: keep target quant (never force Q8_0). Never upscale implicitly.
                _step(f"Fusing style into {tname} (delta + head → {_target_q.name}, kept)…")
                t_quant0 = time.time()
                if _target_q in (Q.F32,):
                    q_head, out_qtype = fused.astype(np.float32), Q.F32
                elif _target_q in (Q.F16,):
                    q_head, out_qtype = fused.astype(np.float16), Q.F16
                else:
                    try:
                        _b = getattr(Q, "BF16", None)
                    except Exception:
                        _b = None
                    if _b is not None and _target_q == _b:
                        q_head, out_qtype = fused.astype(np.float32), Q.F32
                        _warn("  BF16 target: keeping F32 to avoid loss (use --quant bf16 explicitly if needed).")
                        out_qtype = Q.F32
                    else:
                        try:
                            q_head, out_qtype = quantize(fused, _target_q), _target_q
                        except Exception as e:
                            _near = _nearest_encodable(_target_q)
                            if _near is not None:
                                _warn(f"  {_target_q.name} not encodable ({_qerr(e)}); keeping {_near.name} (nearest size) instead of F16.")
                                q_head, out_qtype = _quantize_to_Q(fused, _near)
                            else:
                                _warn(f"  Could not quant to {_target_q.name} ({_qerr(e)}), keeping F16.")
                                q_head, out_qtype = fused.astype(np.float16), Q.F16
                try:
                    _note(f"Fused {fused.nbytes/1e6:.1f} MB → {out_qtype.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
                except Exception:
                    pass
        else:
            # Not delta: voice replace (all quants trivial).
            if _req is not None:
                _step(f"Casting onto {os.path.basename(target)} ({_req.name})…")
                fused = oriented
                t_quant0 = time.time()
                try:
                    q_head, out_qtype = _quantize_to_Q(fused, _req)
                except NotImplementedError as e:
                    _fail(f"  ✗ Quant {_req.name} not implemented by gguf lib ({e}).")
                except Exception as e:
                    _fail(f"  ✗ Could not quant to {_req.name}: {e}")
                _note(f"Quantized {fused.nbytes/1e6:.1f} MB → {_req.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
            elif _is_head:
                # lm_head high-fidelity: quant once to Q8_0 (preserve quality, Q4 saves little).
                if oriented.shape[-1] % 32 != 0:
                    _warn(f"Head last dim {oriented.shape[-1]} not ×32 — Q8_0 needs %32, keeping F16 (larger, compatible).")
                    _step(f"Casting onto {os.path.basename(target)} (F16 fallback)…")
                    fused = oriented
                    q_head, out_qtype = fused.astype(np.float16), Q.F16
                else:
                    _step(f"Casting onto {os.path.basename(target)} (Q8_0)…")
                    fused = oriented
                    t_quant0 = time.time()
                    try:
                        q_head, out_qtype = quantize(fused, Q.Q8_0), Q.Q8_0
                    except Exception as e:
                        sys.stdout.write("\n")
                        if ctx.VERBOSE:
                            _say(f"  Details: {e}")
                        _fail("  ✗ Could not prepare the voiced model. Try again.")
                    _note(f"Quantized {fused.nbytes/1e6:.1f} MB → Q8_0 @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
            else:
                # Generic tensor: keep target quant (fix crash: F32 norms stay F32, never forced Q8_0).
                _tq = _target_q
                try:
                    _tqn = _tq.name
                except Exception:
                    _tqn = str(_tq)
                try:
                    _vqn = vref.dtype
                except Exception:
                    _vqn = "?"
                # Mismatch prompt only for safetensors non-head high-fidelity differing from target.
                _need_prompt = (_vqn.upper() in ("F32", "F16", "BF16", "FP32", "FP16") and _tqn.upper() != _vqn.upper())
                _custom_done = False
                if _need_prompt and _quant_override == "keep" and not _noninteractive(args):
                    try:
                        _say(f"  Target '{tname}' is {_tqn}, fetched is {_vqn}.")
                        _ans = input(f"  Press enter to quantize to {_tqn} or enter custom quant (any like Q8_0/Q4_0/Q6_K/F16/F32/BF16): ").strip()
                    except (EOFError, KeyboardInterrupt):
                        _say("  ⚠ Cancelled.")
                        sys.exit(1)
                    _ans_n = _ans.strip().upper().replace("-", "_")
                    if _ans.strip() == "" or _ans_n == _tqn.upper():
                        pass
                    else:
                        _ck = _parse_quant_override(_ans)
                        _cq = _quant_from_key(_ck) if _ck not in (None, "keep", "target") else None
                        if _ck is None or (_cq is None and _ck not in ("keep", "target")):
                            _warn(f"  Unknown quant '{_ans}', using target {_tqn}.")
                        elif _ck == "target" or _ans_n == _tqn.upper():
                            pass
                        else:
                            _warn(f"Forcing {_ans} for generic tensor (extra quant cycle).")
                            fused = oriented
                            t_quant0 = time.time()
                            try:
                                q_head, out_qtype = _quantize_to_Q(fused, _cq)
                            except NotImplementedError as e:
                                _fail(f"  ✗ Quant {_cq.name} not implemented ({e}).")
                            except Exception as e:
                                _fail(f"  ✗ Could not quant to {_cq.name}: {e}")
                            _note(f"Quantized {fused.nbytes/1e6:.1f} MB → {_cq.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
                            _fd = time.time() - t_fuse0
                            if _fd >= 0.5:
                                _note(f"Fusing done in {_fd:.1f}s")
                            _custom_done = True
                # keep-target path (default, silent when match, Enter=target when mismatch)
                if not _custom_done:
                    _step(f"Casting '{vref.name}' ({_vqn}) onto '{tname}' ({_tqn}, kept)…")
                    fused = oriented
                    t_quant0 = time.time()
                    if _tq in (Q.F32,):
                        q_head, out_qtype = fused.astype(np.float32), Q.F32
                    elif _tq in (Q.F16,):
                        q_head, out_qtype = fused.astype(np.float16), Q.F16
                    else:
                        try:
                            _b = getattr(Q, "BF16", None)
                        except Exception:
                            _b = None
                        if _b is not None and _tq == _b:
                            q_head, out_qtype = fused.astype(np.float32), Q.F32
                            _warn("  BF16 target: keeping F32 to avoid loss (use --quant bf16 explicitly if needed).")
                            out_qtype = Q.F32
                        else:
                            try:
                                q_head, out_qtype = quantize(fused, _tq), _tq
                            except Exception as e:
                                _near = _nearest_encodable(_tq)
                                if _near is not None:
                                    _warn(f"  {_tqn} not encodable ({_qerr(e)}); keeping {_near.name} (nearest size) instead of F16.")
                                    q_head, out_qtype = _quantize_to_Q(fused, _near)
                                else:
                                    q_head, out_qtype = fused.astype(np.float16), Q.F16
                    try:
                        _mbk = fused.nbytes / 1e6
                        _note(f"Kept '{tname}': {_mbk:.1f} MB → {out_qtype.name}" + (f" @ {_mbk / max(time.time() - t_quant0, 1e-6):.0f} MB/s" if fused.nbytes >= 100000 else ""))
                    except Exception:
                        pass
        try:
            t_fuse0
        except Exception:
            t_fuse0 = time.time()
        _fd = time.time() - t_fuse0
        if _fd >= 0.5:
            _note(f"Fusing done in {_fd:.1f}s")
    return q_head, out_qtype


def _gguf_write_repl(target, reader, arch, repl, out_path, args, op, detail=None, _extra_names=()):
    """Single-pass GGUF write. repl maps tensor name -> (array, ggml_type).
    detail (single-tensor callers): (tname, logical, vref, vraw, vqtype, meta) for rich messages."""
    from gguf import GGUFWriter
    from gguf.gguf_writer import GGUFValue
    from gguf.constants import GGMLQuantizationType as Q
    if not arch:
        _warn("  Target architecture unknown — labeling output 'llama' (GGUF requires an arch).")
    writer = GGUFWriter(out_path, arch or "llama", use_temp_file=True)
    for field in reader.fields.values():
        if field.name in ("general.architecture",) or field.name.startswith("GGUF."):
            continue
        try:
            val = field.contents()
        except Exception:
            continue
        vtype = field.types[0] if field.types else None
        if vtype is None:
            continue
        subtype = field.types[-1] if len(field.types) > 1 else None
        try:
            writer.kv_data[0][field.name] = GGUFValue(value=val, type=vtype, sub_type=subtype)
        except Exception:
            try:
                writer.add_key_value(field.name, val, vtype, subtype)
            except Exception:
                if ctx.VERBOSE:
                    _warn(f"Could not copy field {field.name}")

    for t in reader.tensors:
        if t.name in repl:
            _qh, _qt = repl[t.name]
            writer.add_tensor(t.name, _qh, raw_dtype=_qt)
        elif t.tensor_type in (Q.F32, Q.F16):
            writer.add_tensor(t.name, t.data)
        else:
            writer.add_tensor(t.name, t.data, raw_dtype=t.tensor_type)
    for name in _extra_names:
        if name in repl and name not in [t.name for t in reader.tensors]:
            _qh, _qt = repl[name]
            writer.add_tensor(name, _qh, raw_dtype=_qt)

    # writing GGUF is the long wait (often 5-30s for 26GB) — show bar with speed, internet or local
    _step(f"Writing {out_path} …")
    # estimate total bytes for progress — use data.nbytes
    try:
        total_write = 8192
        for t in reader.tensors:
            if t.name in repl:
                continue
            try:
                total_write += int(t.data.nbytes)
            except Exception as e:
                if ctx.VERBOSE:
                    _warn(f"progress estimate for {t.name} failed: {e}")
        try:
            for _qh, _qt in repl.values():
                total_write += int(_qh.nbytes)
        except Exception:
            try:
                for _qh, _qt in repl.values():
                    total_write += int(_qh.size * _qh.itemsize)
            except Exception:
                pass
        if ctx.VERBOSE:
            _note(f"Estimated write {total_write/1e6:.1f} MB ({len(repl)} replaced + {len([t for t in reader.tensors if t.name not in repl])} copied)")
    except Exception as e:
        if ctx.VERBOSE:
            _warn(f"progress estimate failed: {e}")
        total_write = 0
    # ensure total_write is at least header + heads for small models, never 0 for progress
    if total_write < 1000:
        try:
            total_write = 8192
            for _qh, _qt in repl.values():
                try:
                    total_write += int(_qh.nbytes)
                except Exception:
                    pass
            for t in reader.tensors:
                if t.name not in repl:
                    try:
                        total_write += int(t.data.nbytes)
                    except Exception:
                        pass
        except Exception:
            pass
        if total_write < 1000:
            total_write = 0
    t_write0 = time.time()
    last_w = [0.0]
    # progress poll for GGUF write (local disk, but user is waiting)
    import threading
    _stop_poll = [False]
    _tiny_w = bool(total_write) and total_write < 100000
    def _poll_gguf():
        # poll output file (or temp file) size
        target_path = getattr(writer, 'fname', out_path)
        # GGUFWriter with use_temp_file=True writes to temp file with suffix
        # fallback to out_path
        poll_path = target_path if os.path.exists(target_path) else out_path
        # try to find temp file if not yet at final path
        while not _stop_poll[0]:
            try:
                cur = 0
                if os.path.exists(poll_path):
                    cur = os.path.getsize(poll_path)
                else:
                    # try temp file pattern
                    import glob
                    tmps = glob.glob(str(out_path) + "*")
                    for p in tmps:
                        if os.path.isfile(p):
                            try:
                                cur = max(cur, os.path.getsize(p))
                            except Exception:
                                pass
                elapsed = time.time() - t_write0 or 1e-6
                mbps = cur / elapsed / 1e6
                pct = (cur / total_write * 100) if total_write else 0
                if pct > 100:
                    pct = 100
                if total_write:
                    msg = f"  → writing {pct:3.0f}% · {cur/1e6:.1f}/{total_write/1e6:.1f} MB · {mbps:.0f} MB/s  "
                else:
                    msg = f"  → writing {cur/1e6:.1f} MB · {mbps:.0f} MB/s  "
                sys.stdout.write(_transform(msg) + ("\n" if _progress_pref() else "\r"))
                sys.stdout.flush()
            except Exception:
                pass
            time.sleep(0.25)
    _thr = threading.Thread(target=_poll_gguf, daemon=True)
    if not _tiny_w:
        _thr.start()
    try:
        writer.write_header_to_file()
        writer.write_kv_data_to_file()
        writer.write_tensors_to_file()
        writer.close()
    except Exception as e:
        _stop_poll[0] = True
        if not _tiny_w:
            try:
                _thr.join(timeout=0.5)
            except Exception:
                pass
        sys.stdout.write("\n")
        if ctx.VERBOSE:
            _say(f"  Details: {e}")
        _fail(f"  ✗ Could not save the voiced model to {out_path}.")
    _stop_poll[0] = True
    if not _tiny_w:
        try:
            _thr.join(timeout=0.5)
        except Exception:
            pass
        sys.stdout.write("\n")
    elapsed_w = time.time() - t_write0 or 1e-6
    try:
        final_sz = os.path.getsize(out_path)
        _rate_w = f" @ {final_sz / elapsed_w / 1e6:.0f} MB/s" if final_sz >= 100000 else ""
        _note(f"Wrote {final_sz / 1e6:.1f} MB{_rate_w}")
    except Exception:
        pass

    file_size = os.path.getsize(out_path)
    if detail is None:
        summed = [(n, _qt.name if hasattr(_qt, "name") else str(_qt)) for n, (_qh, _qt) in repl.items()]
        log_operation({"op": op, "target": target, "format": "gguf", "tensors": summed,
                       "output": out_path, "status": "ok"})
        _ok(f"Done → {out_path}")
        for n, _qt in summed:
            _say(f"  Cast '{n}' → {_qt} ({file_size/1e6:.1f} MB total).")
        _say(f"  Load it: llama serve -m {out_path}")
        return
    tname, logical, vref, vraw, vqtype, meta = detail
    _qh0, out_qtype = repl[tname]
    log_operation({"op": op, "target": target, "format": "gguf", "tensor": tname,
                   "dtype": out_qtype.name, "shape": list(logical), "output": out_path, "status": "ok"})
    _ok(f"Done → {out_path}")
    is_fallback = 'fallback_to_f16' in locals() and locals().get('fallback_to_f16')
    _is_head_msg = _is_head_name(tname) or _is_head_name(getattr(vref, "name", None))
    _label = "Head" if _is_head_msg else "Tensor"
    if (meta or {}).get("type") == "delta" and is_fallback:
        _say(f"  Style fused and head kept as F16 (Q8_0 needs %32, fallback) ({file_size/1e6:.1f} MB).")
    elif (meta or {}).get("type") == "delta":
        # Delta head keeps Q8_0 for heads, keeps target for generic.
        if _is_head_msg and out_qtype == Q.Q8_0:
            _say(f"  Style fused and head quantized to Q8_0 ({file_size/1e6:.1f} MB).")
        else:
            _say(f"  Style fused into '{tname}' → {out_qtype.name}, kept ({file_size/1e6:.1f} MB).")
    elif is_fallback:
        _say(f"  Head kept as F16 (Q8_0 needs last dim %32, fallback) ({file_size/1e6:.1f} MB).")
    elif out_qtype == Q.Q8_0 and _is_head_msg:
        _qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
        _kept_q8 = (vraw is not None and isinstance(vqtype, str) and vqtype == Q.Q8_0.name and _qo in ("keep",))
        if _kept_q8:
            _say(f"  Head kept as Q8_0 (from {vqtype}, {file_size/1e6:.1f} MB).")
        else:
            _say(f"  Head quantized to Q8_0 ({file_size/1e6:.1f} MB).")
    else:
        _from = vqtype if vraw is not None else getattr(vref, "dtype", "?")
        _say(f"  {_label} '{tname}' kept as {out_qtype.name} (from {_from}, {file_size/1e6:.1f} MB).")
    _say(f"  Load it: llama serve -m {out_path}")


def _read_gguf_arch(reader):
    """Single arch-reader for GGUF targets. Returns the general.architecture string or None."""
    import numpy as np
    for field in reader.fields.values():
        if field.name == "general.architecture":
            try:
                raw = field.parts[field.data[0]]
                return raw.tobytes().decode() if isinstance(raw, np.ndarray) else str(raw)
            except Exception:
                return None
    return None


def _confirm_arch_mismatch(varch, arch, args):
    """Single mismatch warn (+ confirm) shared by single and pack GGUF casts."""
    if varch and arch and varch.lower() != arch.lower():
        _warn(f"This voice was trained for a {varch} model, but the target is {arch}. It may not sound right.")
        if not _confirm("  Continue anyway? [y/N]: ", args):
            _note("Cancelled.")
            sys.exit(1)


def _cast_gguf_speak(target, reader, arch, out_path, vf32, vref, meta, args, op, vraw=None, vqtype=None):
    """--speak: voice onto output.weight, invented if absent. GGUF only."""
    import numpy as np
    from gguf.constants import GGMLQuantizationType as Q
    from gguf.quants import quantize, quant_shape_to_byte_shape
    if (meta or {}).get("type") == "delta":
        _fail("  ✗ --speak cannot place a delta voice (differences are not weights).")
    names = [t.name for t in reader.tensors]
    ot = next((t for t in reader.tensors if t.name == "output.weight"), None)
    et = next((t for t in reader.tensors if t.name == "token_embd.weight"), None)
    if ot is not None:
        olog = tuple(reversed(tuple(int(x) for x in ot.shape)))
        invented = False
    elif et is not None:
        olog = tuple(reversed(tuple(int(x) for x in et.shape)))
        invented = True
    else:
        _fail("  ✗ Target has no output.weight or token_embd.weight to mirror geometry from.")
    varch = (meta or {}).get("architecture")
    _confirm_arch_mismatch(varch, arch, args)
    tname = "output.weight"
    if vraw is not None:
        _vqtype = Q[vqtype]
        if tuple(vref.shape) != olog:
            _fail(f"  ✗ Voice shape {vref.shape} doesn't fit {tname} {olog}.")
        _qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
        if _qo not in ("keep",):
            fitted = _gguf_raw_to_f32(vraw, _vqtype, tuple(reversed(tuple(int(x) for x in vref.shape))))
            q_head, out_qtype = _quantize_to_Q(np.ascontiguousarray(fitted), Q[_qo.upper()])
        else:
            _need = 1
            for _d in quant_shape_to_byte_shape(tuple(int(x) for x in vref.shape), _vqtype):
                _need *= _d
            _raw = np.frombuffer(vraw, dtype=np.uint8)
            if _raw.size < _need:
                _fail(f"  ✗ Voice data too short for {vqtype} {tuple(vref.shape)}.")
            q_head = np.ascontiguousarray(_raw[:_need].reshape(
                quant_shape_to_byte_shape(tuple(int(x) for x in vref.shape), _vqtype)))
            out_qtype = _vqtype
    else:
        fitted = _orient(vf32, vref.shape, olog)
        if fitted is None:
            _fail(f"  ✗ Voice shape {vref.shape} doesn't fit {tname} {olog}.")
        fitted = np.ascontiguousarray(fitted)
        _qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
        if _qo is None:
            _fail(f"  ✗ Invalid --quant '{getattr(args, 'quant', None)}'.")
        if _qo not in ("keep",):
            q_head, out_qtype = _quantize_to_Q(fitted, Q[_qo.upper()])
        elif fitted.shape[-1] % 32 == 0:
            q_head, out_qtype = quantize(fitted, Q.Q8_0), Q.Q8_0
        else:
            _note(f"  Head last dim {fitted.shape[-1]} not ×32 — Q8_0 needs %32, keeping F16 (larger, compatible).")
            q_head, out_qtype = fitted.astype(np.float16), Q.F16
    if invented:
        _note("  speak: voice → invented output.weight (file untied)")
    else:
        _note("  speak: voice → output.weight")
    _gguf_write_repl(target, reader, arch, {tname: (q_head, out_qtype)}, out_path, args, op,
                      detail=(tname, olog, vref, vraw, vqtype, meta),
                      _extra_names=(tname,) if invented else ())

def _cast_gguf(target, out_path, vf32, vref, meta, args, op, vraw=None, vqtype=None, strict_target=False, target_name=None):
    import numpy as np
    from gguf import GGUFReader

    reader = GGUFReader(target)
    arch = _read_gguf_arch(reader)

    if getattr(args, "speak", False):
        _cast_gguf_speak(target, reader, arch, out_path, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype)
        return
    target_tensor = None
    if target_name is not None:
        # Explicitly routed (requested subset): honor it, never fall back.
        for t in reader.tensors:
            if t.name == target_name:
                target_tensor = t
                break
        if target_tensor is None:
            _fail(f"  ✗ Resolved target '{target_name}' for '{vref.name}' not in {target}.")
    if target_tensor is None and vref.name:
        for t in reader.tensors:
            if t.name == vref.name:
                target_tensor = t
                break
    if target_tensor is None and strict_target:
        _fail(f"  ✗ Tensor '{vref.name}' not found in target {target}. Voicepack requires exact match (not mapping to another tensor).")
    if target_tensor is None:
        for preferred in ("output.weight", "token_embd.weight"):
            for t in reader.tensors:
                if t.name == preferred:
                    target_tensor = t
                    break
            if target_tensor is not None:
                break
    if target_tensor is None:
        _fail("  ✗ Target has no output.weight or token_embd.weight to receive the voice.")

    tname = target_tensor.name
    logical = tuple(reversed(tuple(int(x) for x in target_tensor.shape)))

    # arch mismatch warning
    varch = (meta or {}).get("architecture")
    _confirm_arch_mismatch(varch, arch, args)

    q_head, out_qtype = _gguf_fuse_one(target, target_tensor, tname, logical, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype)
    _gguf_write_repl(target, reader, arch, {tname: (q_head, out_qtype)}, out_path, args, op,
                      detail=(tname, logical, vref, vraw, vqtype, meta))


# GGUF (llama.cpp) <-> safetensors (HF) tensor-name aliases, matched by
# (block index, role). Exact name match always wins; aliases only fire on a
# miss, and only when exactly one candidate matches. Anything else fails loud
# at the call site — never a silent mapping to another tensor.
_ROLE_ALIASES = (
    ("embed_tokens.weight", "token_embd.weight"),
    ("word_embeddings.weight", "token_embd.weight"),
    ("wte.weight", "token_embd.weight"),
    ("output_layer.weight", "output.weight"),
    ("lm_head.weight", "output.weight"),
    ("final_layernorm.weight", "output_norm.weight"),
    ("self_attention.dense.weight", "attn_output.weight"),
    ("mlp.dense_4h_to_h.weight", "ffn_down.weight"),
    ("model.norm.weight", "output_norm.weight"),
    ("ln_f.weight", "output_norm.weight"),
    ("input_layernorm.weight", "attn_norm.weight"),
    ("self_attn.q_proj.weight", "attn_q.weight"),
    ("self_attn.k_proj.weight", "attn_k.weight"),
    ("self_attn.v_proj.weight", "attn_v.weight"),
    ("self_attn.o_proj.weight", "attn_o.weight"),
    ("self_attn.o_proj.weight", "attn_output.weight"),
    ("post_attention_layernorm.weight", "post_attention_norm.weight"),
    ("mlp.down_proj.weight", "ffn_down.weight"),
    ("experts.down_proj", "ffn_down_exps.weight"),
    ("mlp.up_proj.weight", "ffn_up.weight"),
)


def _block_index(name):
    import re
    m = re.search(r"(?<![A-Za-z0-9_])(?:layers|layer|blk|blks|blocks|h)\.(\d+)\b", name)
    if m:
        return int(m.group(1))
    m = re.search(r"\.(\d+)\.", name)
    return int(m.group(1)) if m else None


def _pack_aliases(vname, target_names):
    """All alias matches for vname in target_names (either naming direction).
    Exact match is handled by callers first; this only fires on a miss."""
    hits = []
    for st_suf, gg_suf in _ROLE_ALIASES:
        for t in target_names:
            st_v = vname.endswith(st_suf)
            gg_v = vname.endswith(gg_suf)
            st_t = t.endswith(st_suf)
            gg_t = t.endswith(gg_suf)
            if (st_v and gg_t) or (gg_v and st_t):
                if _block_index(vname) == _block_index(t) and t not in hits:
                    hits.append(t)
    return hits


def _role_key(name):
    """(block_index, trailing_role) canonical key for same-format matching.
    Trailing role is everything after the block number (e.g.
    'self_attn.o_proj.weight'); unblocked names fall back to their last two
    components (e.g. 'embed_tokens.weight'). The .weight/.bias suffix stays
    in the key so a bias never matches a weight. The block keyword
    (layers/blk/...) is left-guarded against mid-word hits; the generic
    dotted-number fallback still fires on version-like segments, so the
    elem-count gate remains the real false-friend veto."""
    import re
    m = re.search(r"(?<![A-Za-z0-9_])(?:layers|layer|blk|blks|blocks|h)\.(\d+)\b", name)
    if m:
        return (int(m.group(1)), name[m.end():].strip("."))
    m = re.search(r"\.(\d+)\.", name)
    if m:
        return (int(m.group(1)), name[m.end():].strip("."))
    parts = name.split(".")
    return (None, ".".join(parts[-2:]))


def _same_role(vname, target_names):
    """Targets sharing vname's (block, trailing role). Cross-format pairs
    (ST suffix vs GGUF suffix) never collide here — _ROLE_ALIASES owns those.
    Only fires on a miss; callers gate on elem-count and uniqueness."""
    want = _role_key(vname)
    return [t for t in target_names if _role_key(t) == want]


def _cast_gguf_pack(target, out_path, items, args, op):
    """Single-pass pack cast onto GGUF. items: [(vf32, vref, meta, vraw, vqtype)]."""
    import numpy as np
    from gguf import GGUFReader
    reader = GGUFReader(target)
    arch = _read_gguf_arch(reader)
    varch = (items[0][2] or {}).get("architecture") if items else None
    _confirm_arch_mismatch(varch, arch, args)
    _names = [t.name for t in reader.tensors]
    _counts = {}
    for t in reader.tensors:
        try:
            n = 1
            for d in t.shape:
                n *= int(d)
            _counts[t.name] = n
        except Exception:
            pass
    _map = _resolve_pack_targets([(ref.name, ref.shape) for (_, ref, _, _, _) in items],
                                 _names, target, _counts)
    repl = {}
    for vf32, vref, meta, vraw, vqtype in items:
        target_tensor = next(t for t in reader.tensors if t.name == _map[vref.name])
        tname = target_tensor.name
        logical = tuple(reversed(tuple(int(x) for x in target_tensor.shape)))
        q_head, out_qtype = _gguf_fuse_one(target, target_tensor, tname, logical, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype)
        repl[tname] = (q_head, out_qtype)
    _gguf_write_repl(target, reader, arch, repl, out_path, args, op)


def _resolve_pack_targets(pairs, target_names, target, counts=None):
    """Map every voice tensor to a target tensor (exact, else one alias,
    else one same-role same-format match).
    Resolves ALL up front so a pack fails once with every miss — never
    fix-forward one at a time. Same-size targets are offered as hints only.
    pairs: [(voice_name, voice_shape)]; counts: {target_name: elem_count}.
    Voice shapes are logical-as-is (GgufSource.ref already reverses);
    only elem-counts are compared, so transposes pass and nothing double-flips."""
    def _want(vshape):
        if vshape is None:
            return None
        n = 1
        for d in vshape:
            n *= int(d)
        return n
    mapping, misses = {}, []
    for vname, vshape in pairs:
        if vname in target_names:
            mapping[vname] = vname
            continue
        cands = _pack_aliases(vname, target_names)
        if len(cands) == 1:
            mapping[vname] = cands[0]
            _note(f"  Aliased '{vname}' → '{cands[0]}' ({target} naming).")
        elif len(cands) > 1:
            misses.append(f"  '{vname}' is ambiguous: {', '.join(cands)}")
        else:
            roles = [t for t in _same_role(vname, target_names)
                     if _want(vshape) is None or counts is None
                     or counts.get(t, _want(vshape)) == _want(vshape)]
            if len(roles) == 1:
                mapping[vname] = roles[0]
                _note(f"  Aliased '{vname}' → '{roles[0]}' (same role).")
                continue
            if len(roles) > 1:
                misses.append(f"  '{vname}' is ambiguous: {', '.join(roles)}")
                continue
            hint = ""
            if counts is not None and vshape is not None:
                same = [t for t in target_names if counts.get(t) == _want(vshape)][:3]
                if same:
                    hint = f" (same size: {', '.join(same)})"
            misses.append(f"  '{vname}' not found{hint}")
    if misses:
        _fail(f"  ✗ {len(misses)} tensor(s) have no match in target {target}:\n"
              + "\n".join(misses)
              + "\n  Voicepack requires exact match (not mapping to another tensor).")
    return mapping


def _cast_pack(target, src_path, out_path, args, op, excludes=(), meta=None, order=()):
    """Cast every tensor in a pack file onto target (minus excludes).
    GGUF and safetensors targets each get one write pass (repl-dict fused first).
    Delta marker on the pack is detected once and threaded into every tensor.
    order (requested subset): process in the user's order, not file order."""
    if getattr(args, "speak", False) and not _is_gguf_file(target):
        _fail("  ✗ --speak needs a .gguf target. Re-run without --speak for .safetensors, or point cast at a .gguf file.")
    src = open_source(str(src_path))
    all_names = [n for n in src.names() if n != "delta.voice.marker"]
    for e in excludes:
        if e not in all_names and not any(e in n for n in all_names):
            _warn(f"  --exclude '{e}' not in pack (has {len(all_names)} tensors). Ignoring.")

    resolved_ex = set()
    for e in excludes:
        if e in all_names:
            resolved_ex.add(e)
        else:
            _m = [n for n in all_names if e in n]
            if len(_m) == 1:
                resolved_ex.add(_m[0])
            elif len(_m) > 1:
                _fail(f"  ✗ --exclude '{e}' is ambiguous: {', '.join(_m)}. Name one tensor exactly.")
    to_cast = [n for n in all_names if n not in resolved_ex]
    if order:
        _pos = {n: i for i, n in enumerate(order)}
        to_cast.sort(key=lambda n: (_pos.get(n, len(_pos)), n))
    if not to_cast:
        _fail("  ✗ --exclude removes all tensors. Nothing to cast.")
    _step(f"Casting {len(to_cast)} tensors from {Path(src_path).name} onto {Path(target).name}…")
    if resolved_ex:
        _note(f"Excluding {len(resolved_ex)}: {', '.join(sorted(resolved_ex))}")
    is_delta_file = str(src_path).endswith(".safetensors") and _is_delta_safetensors(str(src_path))
    if is_delta_file:
        _note("Detected delta voice — fusing style (delta + head)…")
    if meta is None and is_delta_file:
        meta = {"type": "delta", "source": str(src_path)}
    items = []
    for i, tname in enumerate(to_cast):
        f32, ref, raw, qtype = _load_tensor(src_path, tname, _src=src)
        _step(f"[{i+1}/{len(to_cast)}] {tname} ({ref.dtype} {tuple(ref.shape)})…")
        items.append((f32, ref, meta, raw, qtype))
    if _is_gguf_file(target):
        _cast_gguf_pack(target, out_path, items, args, op)
        return
    _header, _tensors = read_safetensors(target)
    _st_names = list(_tensors.keys())
    _st_counts = {}
    for _n, _info in _tensors.items():
        try:
            _c = 1
            for _d in _info["shape"]:
                _c *= int(_d)
            _st_counts[_n] = _c
        except Exception:
            pass
    _st_map = _resolve_pack_targets([(ref.name, ref.shape) for (_, ref, _, _, _) in items],
                                    _st_names, target, _st_counts)
    repl = {}
    for f32, ref, m, raw, qtype in items:
        tname, data, tdtype, tshape = _cast_st_prepare(target, _tensors, f32, ref, m, args,
                                                      vraw=raw, vqtype=qtype, strict_target=True,
                                                      target_name=_st_map[ref.name])
        repl[tname] = (data, tdtype, tshape)
    _step(f"Replacing {len(repl)} tensors in {Path(target).name}…")
    _safetensors_write_repl(target, repl, out_path)
    file_size = os.path.getsize(out_path)
    log_operation({"op": op, "target": target, "format": "safetensors",
                   "tensors": [(n, d) for n, (_b, d, _s) in repl.items()],
                   "output": out_path, "status": "ok"})
    _ok(f"Done → {out_path}")
    for n, (_b, d, _s) in repl.items():
        _say(f"  Tensor '{n}' replaced at {d} ({file_size/1e6:.1f} MB total).")


def _parse_tensor_list(args):
    # Tensor positionals (get/cast tensors...), with comma-split for shell convenience.
    raw = getattr(args, "tensors", None) or []
    if isinstance(raw, str):
        raw = [raw]
    # Split comma-separated entries, strip, drop empties.
    out = []
    for r in raw:
        if r is None:
            continue
        for part in str(r).split(","):
            part = part.strip()
            if part:
                out.append(part)
    return out


def _parse_exclude_list(args):
    raw = getattr(args, "excludes", None) or getattr(args, "exclude", None) or []
    if isinstance(raw, str):
        raw = [raw]
    out = []
    for r in (raw or []):
        for part in str(r).split(","):
            part = part.strip()
            if part:
                out.append(part)
    return out


def _resolve_one_tensor(req, names):
    # Exact match first, then single substring match, else fail. Returns resolved name.
    if req in names:
        return req
    matches = [n for n in names if req and req in n]
    if len(matches) == 1:
        return matches[0]
    if len(matches) > 1:
        _fail(f"  ✗ '{req}' is ambiguous: {', '.join(matches[:8])}…")
    _fail(f"  ✗ Tensor '{req}' not found. Try one of: {', '.join(names[:10])}")


def _tensor_src_meta(src_path):
    """File-level delta detection for the tensor path (the marker lives in the file; no voice.json)."""
    if str(src_path).endswith(".safetensors"):
        try:
            if _is_delta_safetensors(str(src_path)):
                return {"type": "delta", "source": str(src_path)}
        except Exception:
            pass
    return None


def _load_tensor(path, name=None, _src=None):
    src = _src if _src is not None else open_source(str(path))
    names = src.names()
    if not names:
        _fail(f"  ✗ No tensors found in {path}.")
    if name is None:
        name = "output.weight" if "output.weight" in names else names[0]

    if name not in names:
        matches = [n for n in names if name in n]
        if len(matches) == 1:
            name = matches[0]
        else:
            _fail(f"  ✗ Tensor '{name}' not found in {path}.")
    ref = src.ref(name)
    if src.kind == "gguf" and _is_ggml_quant(ref.dtype):
        return None, ref, src.read_raw(name), ref.dtype
    return src.read_f32(name), ref, None, None