tests/test_units.py

"""Offline unit tests: pure logic, no network. Must stay fast (<60s total)."""
import json
import os
import shutil
import struct
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _helpers import IsolatedVoices, ns, voice  # noqa: E402
from vlib.pack import _resolve_pack_tensors  # noqa: E402
from vlib import della as voice_della  # noqa: E402

import numpy as np


def write_st(path, tensors):
    header, off = {}, 0
    for name, (arr, dt) in tensors.items():
        b = arr.tobytes()
        header[name] = {"dtype": dt, "shape": list(arr.shape),
                        "data_offsets": [off, off + len(b)]}
        off += len(b)
    hj = json.dumps(header).encode()
    Path(path).write_bytes(struct.pack("<Q", len(hj)) + hj
                           + b"".join(t[0].tobytes() for t in tensors.values()))


def read_arr(path, name, dtype=np.float16):
    raw = Path(path).read_bytes()
    hs = struct.unpack("<Q", raw[:8])[0]
    hdr = json.loads(raw[8:8 + hs])
    info = hdr[name]
    b = raw[8 + hs + info["data_offsets"][0]:8 + hs + info["data_offsets"][1]]
    return np.frombuffer(b, dtype=dtype).reshape(info["shape"])

def _mktmp(self):
    d = Path(tempfile.mkdtemp())
    self.addCleanup(shutil.rmtree, d, True)
    return d


class TestNames(unittest.TestCase):
    def test_sanitize_accepts_sane(self):
        for good in ("a", "ok-name_1", "StyleTune-V2", "x" * 64):
            self.assertIsNotNone(voice._sanitize_voice_name(good), good)

    def test_sanitize_rejects(self):
        for bad in ("", "  ", "my voice", "a/b", "a.b", "a:b", "delta-foo", ".", "..", "a\nb"):
            self.assertIsNone(voice._sanitize_voice_name(bad), repr(bad))

    def test_slugify(self):
        self.assertEqual(voice._slugify_voice_name("my model"), "my-model")
        self.assertEqual(voice._slugify_voice_name("My.Model-v2"), "my-model-v2")
        self.assertEqual(voice._slugify_voice_name(""), "voice")
        self.assertIsNotNone(voice._sanitize_voice_name(voice._slugify_voice_name("WEIRD name!!")))

    def test_resolve_name_uses_slug_for_defaults(self):
        self.assertEqual(voice._resolve_name(ns(name=None), "my model"), "my-model")

    def test_resolve_name_rejects_explicit(self):
        with self.assertRaises(SystemExit):
            voice._resolve_name(ns(name="bad name!"), "fallback")

    def test_public_source_passthrough(self):
        for good in ("org/model", "google/gemma-4-12b", "model.gguf",
                     "voice-name", "", "—"):
            self.assertEqual(voice._public_source(good), good, good)

    def test_public_source_strips_local_paths(self):
        self.assertEqual(
            voice._public_source("/Users/someone/Downloads/m.gguf"), "m.gguf")
        self.assertEqual(voice._public_source("./rel/m.gguf"), "m.gguf")
        self.assertEqual(voice._public_source("~/m.gguf"), "m.gguf")

    def test_public_source_strips_url_secrets(self):
        self.assertEqual(
            voice._public_source("https://user:pw@huggingface.co/o/m"),
            "https://huggingface.co/o/m")
        self.assertEqual(
            voice._public_source("https://cdn.example/x.bin?sig=SECRET&t=1"),
            "https://cdn.example/x.bin")

    def test_registry_folder_lowercases(self):
        self.assertTrue(str(voice._registry_folder("StyleTune-V2")).endswith("styletune-v2"))


class TestDtypes(unittest.TestCase):
    def test_roundtrips(self):
        arr = (np.arange(24, dtype=np.float32).reshape(4, 6) - 12) / 3
        for dt, tol in (("F32", 0.0), ("F16", 1e-3), ("BF16", 2e-2)):
            back = voice.decode_to_f32(voice.encode_from_f32(arr, dt), dt, (4, 6))
            self.assertLessEqual(float(np.abs(arr - back).max()), tol, dt)

    def test_bf16_rounds_not_truncates(self):
        rng = np.random.default_rng(0)
        arr = rng.normal(size=(64, 64)).astype(np.float32)
        u32 = arr.view(np.uint32)
        trunc = ((u32 >> 16).astype(np.uint16).astype(np.uint32) << 16).view(np.float32)
        back = voice.decode_to_f32(voice.encode_from_f32(arr, "BF16"), "BF16", arr.shape)
        self.assertLessEqual(float(np.abs(arr - back).max()),
                             float(np.abs(arr - trunc).max()) + 1e-7)

    def test_rejects_garbage(self):
        with self.assertRaises(ValueError):
            voice.dtype_bytes("Q4_0")
        with self.assertRaises(ValueError):
            voice.decode_to_f32(b"\x00" * 4, "F32", (2,))
        with self.assertRaises(ValueError):
            voice.encode_from_f32(np.zeros(4, dtype=np.float32), "Q8_0")


class TestHeaders(unittest.TestCase):
    def test_validate_catches(self):
        good = {"a": {"dtype": "F32", "shape": [2], "data_offsets": [0, 8]}}
        voice._validate_safetensors_header(dict(good), 100, 50)
        with self.assertRaises(ValueError):  # overlap
            voice._validate_safetensors_header(
                {"a": {"dtype": "F32", "shape": [2], "data_offsets": [0, 8]},
                 "b": {"dtype": "F32", "shape": [2], "data_offsets": [4, 12]}}, 100, 50)
        with self.assertRaises(ValueError):  # size mismatch
            voice._validate_safetensors_header(
                {"a": {"dtype": "F32", "shape": [4], "data_offsets": [0, 8]}}, 100, 50)
        with self.assertRaises(ValueError):  # unknown dtype
            voice._validate_safetensors_header(
                {"a": {"dtype": "Q4_0", "shape": [2], "data_offsets": [0, 8]}}, 100, 50)
        with self.assertRaises(ValueError):  # out of bounds
            voice._validate_safetensors_header(
                {"a": {"dtype": "F32", "shape": [2], "data_offsets": [0, 999]}}, 100, 50)

    def test_write_read_roundtrip(self):
        p = _mktmp(self) / "t.safetensors"
        write_st(p, {"output.weight": (np.ones((4, 4), dtype=np.float16), "F16")})
        hdr, tens = voice.read_safetensors(str(p))
        self.assertIn("output.weight", tens)

    def test_rewrite_preserves_others_and_guards_size(self):
        d = _mktmp(self)
        tgt, out = d / "t.safetensors", d / "o.safetensors"
        write_st(tgt, {"output.weight": (np.zeros((4, 4), dtype=np.float16), "F16"),
                       "a.weight": (np.full((2, 2), 3.0, dtype=np.float16), "F16")})
        voice.rewrite_safetensors_tensor(str(tgt), str(out), "output.weight",
                                         np.ones((4, 4), dtype=np.float16).tobytes(),
                                         "F16", (4, 4))
        self.assertTrue((read_arr(out, "output.weight") == 1.0).all())
        self.assertTrue((read_arr(out, "a.weight") == 3.0).all())
        with self.assertRaises(SystemExit):
            voice.rewrite_safetensors_tensor(str(tgt), str(d / "bad.safetensors"),
                                             "output.weight", b"\x00" * 2, "F16", (1, 1))


class TestDetect(unittest.TestCase):
    def test_exact_alias_wins(self):
        self.assertEqual(voice.detect_output_tensor(["a", "lm_head.weight"], {}), "lm_head.weight")

    def test_substring_alias(self):
        n = "transformer.output_layer.weight"
        self.assertEqual(voice.detect_output_tensor(["other", n], {}), n)

    def test_tied_falls_back_to_embed(self):
        self.assertEqual(voice.detect_output_tensor(
            ["model.embed_tokens.weight"], {"tie_word_embeddings": True}),
            "model.embed_tokens.weight")

    def test_untied_embed_is_last_resort(self):
        self.assertEqual(voice.detect_output_tensor(["model.embed_tokens.weight"], {}),
                         "model.embed_tokens.weight")
        self.assertIsNone(voice.detect_output_tensor(["blk.0.weight"], {}))

    def test_token_embd_detected_as_head(self):
        self.assertEqual(voice.detect_output_tensor(["blk.0.weight", "token_embd.weight"], None), "token_embd.weight")
        self.assertEqual(voice.detect_output_tensor(["output.weight", "token_embd.weight"], None), "output.weight")

    def test_largest_2d(self):
        self.assertEqual(voice.largest_2d(["a", "b"], {"a": (10, 20), "b": (5, 5)}), "a")

    def test_gguf_prefers_alias_over_size(self):
        class FakeGGUF:
            kind = "gguf"

            def ref(self, n):
                shapes = {"some.huge.weight": (64, 64),
                          "transformer.output_layer.weight": (32, 32)}
                return voice.TensorRef(n, "F16", shapes[n])

        names = ["some.huge.weight", "transformer.output_layer.weight"]
        self.assertEqual(voice._resolve_output_name(FakeGGUF(), None, names),
                         "transformer.output_layer.weight")


class TestOrient(unittest.TestCase):
    def test_same_transpose_miss(self):
        a = np.ones((2, 3), dtype=np.float32)
        self.assertIs(voice._orient(a, (2, 3), (2, 3)), a)
        self.assertEqual(voice._orient(a, (2, 3), (3, 2)).shape, (3, 2))
        self.assertIsNone(voice._orient(a, (2, 3), (4, 4)))


class TestQuantNets(unittest.TestCase):
    def test_parse_quant(self):
        self.assertEqual(voice._parse_quant_override(None), "keep")
        self.assertEqual(voice._parse_quant_override("Q4_0"), "q4_0")
        self.assertEqual(voice._parse_quant_override("fp16"), "f16")
        self.assertIsNone(voice._parse_quant_override("bogus-quant"))

    def test_retry_after_bounded(self):
        self.assertEqual(voice._bounded_retry_after({"Retry-After": "3600"}, 1.0), 30)
        self.assertEqual(voice._bounded_retry_after({"Retry-After": "2"}, 1.0), 2)
        self.assertEqual(voice._bounded_retry_after({}, 1.5), 1.5)

    def test_hf_url_table(self):
        self.assertTrue(voice._is_hf_url("https://huggingface.co/o/m"))
        self.assertTrue(voice._is_hf_url("https://cdn-lfs.huggingface.co/x"))
        self.assertFalse(voice._is_hf_url("https://us.aws.cdn.hf.co/xet-bridge/x"))
        self.assertFalse(voice._is_hf_url("https://evil.com/huggingface.co/x"))

    def test_auth_never_leaks_off_host(self):
        h = voice._hf_auth_headers("https://evil.com/x", {"Authorization": "Bearer S"})
        self.assertNotIn("Authorization", h)


class TestGuards(IsolatedVoices):
    def test_check_overwrite_refuses_symlink(self):
        d = _mktmp(self)
        real = d / "real.gguf"
        real.write_bytes(b"x")
        link = d / "link.gguf"
        try:
            link.symlink_to(real)
        except OSError:
            self.skipTest("symlinks unsupported here")
        with self.assertRaises(SystemExit):
            voice._check_overwrite(str(link), ns())

    def test_assert_not_symlink(self):
        d = _mktmp(self)
        link = d / "l"
        try:
            link.symlink_to(d / "t")
        except OSError:
            self.skipTest("symlinks unsupported here")
        with self.assertRaises(SystemExit):
            voice._assert_not_symlink(str(link))


class TestAddCache(IsolatedVoices):
    def test_add_tensor_before_metadata(self):
        src = _mktmp(self)
        write_st(src / "voice.safetensors",
                 {"output.weight": (np.zeros((4, 4), dtype=np.float16), "F16")})
        (src / "voice.json").write_text(json.dumps({"name": "myadd", "source": "t"}))
        voice.cmd_add(ns(safetensors=str(src / "voice.safetensors"),
                         json=str(src / "voice.json"), name="myadd"))
        self.assertTrue((self.vdir / "myadd" / "voice.safetensors").exists())
        self.assertTrue((self.vdir / "myadd" / "voice.json").exists())

    def test_add_invalid_leaves_no_metadata(self):
        src = _mktmp(self)
        (src / "bad.safetensors").write_bytes(b"corrupt-not-safetensors")
        (src / "voice.json").write_text(json.dumps({"name": "badadd"}))
        with self.assertRaises(SystemExit):
            voice.cmd_add(ns(safetensors=str(src / "bad.safetensors"),
                             json=str(src / "voice.json"), name="badadd"))
        self.assertFalse((self.vdir / "badadd" / "voice.json").exists())

    def test_cache_base_survives_tmp_cleanup(self):
        src = _mktmp(self)
        write_st(src / "base.safetensors",
                 {"output.weight": (np.zeros((4, 4), dtype=np.float16), "F16")})
        voice._cache_base(str(src / "base.safetensors"),
                          self.vdir / "base" / "m.safetensors",
                          self.vdir / "base" / "m.json")
        self.assertTrue((self.vdir / "base" / "m.safetensors").exists())
        meta = json.loads((self.vdir / "base" / "m.json").read_text())
        self.assertEqual(meta.get("bytes"), 4 * 4 * 2)


class TestDeltaGates(unittest.TestCase):
    def _pair(self, d, marker_voice=False, marker_base=False):
        import json as _j
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"output.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"output.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16")})
        for p, mk in ((v, marker_voice), (b, marker_base)):
            if not mk:
                continue
            raw = p.read_bytes()
            hs = struct.unpack("<Q", raw[:8])[0]
            body = raw[8 + hs:]
            total = 4 * 4 * 2
            hdr = {"output.weight": {"dtype": "F16", "shape": [4, 4],
                                     "data_offsets": [0, total]},
                   "delta.voice.marker": {"dtype": "U8", "shape": [1],
                                          "data_offsets": [total, total + 1]}}
            hj = _j.dumps(hdr).encode()
            p.write_bytes(struct.pack("<Q", len(hj)) + hj + body[:total] + b"\x01")
        return v, b

    def test_normal_delta_computes(self):
        d = _mktmp(self)
        v, b = self._pair(d)
        out = voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())
        got = read_arr(out, "output.weight")
        self.assertTrue((got == 1.0).all(), "delta must equal voice(2.0) - base(1.0)")

    def test_tied_base_head_falls_back_to_embed(self):
        import io
        from contextlib import redirect_stdout
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"lm_head.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16"),
                     "model.language_model.embed_tokens.weight": (np.full((4, 4), 7.0, dtype=np.float16), "F16")})
        write_st(b, {"model.language_model.embed_tokens.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16")})
        buf = io.StringIO()
        with redirect_stdout(buf):
            out = voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())
        self.assertTrue((read_arr(out, "lm_head.weight") == 1.0).all(), "head delta must equal voice head(2.0) - tied base embed(1.0)")
        self.assertTrue((read_arr(out, "model.language_model.embed_tokens.weight") == 6.0).all(), "embed delta must equal voice embed(7.0) - base embed(1.0)")
        self.assertIn("ties its head", buf.getvalue(), "fallback must be loud, not silent")

    def test_tied_base_shape_mismatch_fails(self):
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"lm_head.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"model.language_model.embed_tokens.weight": (np.full((4, 8), 1.0, dtype=np.float16), "F16")})
        with self.assertRaises(SystemExit):
            voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())

    def test_headless_baseless_still_fails(self):
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"lm_head.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"blk.0.attn.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16")})
        with self.assertRaises(SystemExit):
            voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())

    def test_resolve_head_want_table(self):
        E = "model.language_model.embed_tokens.weight"
        F = lambda: (np.full((4, 4), 1.0, dtype=np.float16), "F16")
        cases = [
            ("reroutes tied head to embed",
             {E: F()}, ["lm_head.weight"], [E]),
            ("dedupes embed",
             {E: F()}, ["lm_head.weight", E], [E]),
            ("untied unchanged",
             {"lm_head.weight": F(), E: F()}, ["lm_head.weight"], ["lm_head.weight"]),
            ("headless unchanged",
             {"blk.0.attn.weight": F()}, ["lm_head.weight"], ["lm_head.weight"]),
            ("cross-format alias",
             {"lm_head.weight": F()}, ["output.weight"], ["lm_head.weight"]),
        ]
        for label, tensors, want, expected in cases:
            with self.subTest(label):
                d = _mktmp(self)
                b = d / "b.safetensors"
                write_st(b, tensors)
                self.assertEqual(voice._resolve_head_want(str(b), None, want), expected)

    def test_role_table(self):
        self.assertEqual(voice.role_of("lm_head.weight"), "head")
        self.assertEqual(voice.role_of("model.output.weight"), "head")
        self.assertEqual(voice.role_of("token_embd.weight"), "embed")
        self.assertEqual(voice.role_of("model.language_model.embed_tokens.weight"), "embed")
        self.assertEqual(voice.role_of("blk.0.ffn_down.weight"), "body")
        self.assertEqual(voice.role_of(""), "body")
        self.assertTrue(voice.is_tied(["model.language_model.embed_tokens.weight", "blk.0.weight"]))
        self.assertTrue(voice.is_tied(["token_embd.weight"]))
        self.assertFalse(voice.is_tied(["lm_head.weight", "model.language_model.embed_tokens.weight"]))
        self.assertFalse(voice.is_tied(["output.weight", "token_embd.weight"]))
        self.assertFalse(voice.is_tied([]))
        self.assertEqual(voice.find_head(["blk.0.weight", "output.weight"]), "output.weight")
        self.assertIsNone(voice.find_head(["blk.0.weight", "token_embd.weight"]))
        self.assertEqual(voice.find_embed(["blk.0.weight", "token_embd.weight"]), "token_embd.weight")

    def test_cross_format_head_delta_computes(self):
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"output.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"lm_head.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16"),
                     "model.language_model.embed_tokens.weight": (np.full((4, 4), 9.0, dtype=np.float16), "F16")})
        out = voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())
        self.assertTrue((read_arr(out, "output.weight") == 1.0).all(), "cross-format head delta must equal voice(2.0) - base(1.0)")

    def test_gguf_tied_base_delta_computes(self):
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"lm_head.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"token_embd.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16"),
                     "blk.0.attn.weight": (np.full((4, 4), 5.0, dtype=np.float16), "F16")})
        out = voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())

    def test_ambiguous_head_fails_loud(self):
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"model.lm_head.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"lm_head.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16"),
                     "output.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16")})
        with self.assertRaises(SystemExit):
            voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())

    def test_gguf_voice_to_st_base_cross_format(self):
        from gguf import GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        vp = d / "v.gguf"
        w = GGUFWriter(str(vp), "llama", use_temp_file=False)
        w.add_tensor("output.weight", np.full((4, 4), 2.0, dtype=np.float16), raw_dtype=Q.F16)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        b = d / "b.safetensors"
        write_st(b, {"lm_head.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16"),
                     "model.language_model.embed_tokens.weight": (np.full((4, 4), 9.0, dtype=np.float16), "F16")})
        out = voice._compute_delta(str(vp), str(b), d / "out", "delta-x", ns())
        self.assertTrue((read_arr(out, "output.weight") == 1.0).all(), "GGUF voice head must delta against ST base head")

    def test_gguf_delta_chunked_matches_full(self):
        # Slow path (either side GGUF) must stream dequant in bounded chunks:
        # patch the seam small + unaligned, spy max slice, assert values equal
        # full-dequant math and every slice bounded by the seam.
        from gguf import GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        from gguf.quants import quantize
        d = _mktmp(self)
        shape = (64, 128)
        rng = np.random.default_rng(7)
        vf = (rng.normal(size=shape) * 2).astype(np.float32)
        bf = (rng.normal(size=shape) * 2).astype(np.float32)
        vp, bp = d / "v.gguf", d / "b.gguf"
        for p, f in ((vp, vf), (bp, bf)):
            w = GGUFWriter(str(p), "llama", use_temp_file=False)
            w.add_tensor("output.weight", quantize(f, Q.Q8_0), raw_dtype=Q.Q8_0)
            w.write_header_to_file()
            w.write_kv_data_to_file()
            w.write_tensors_to_file()
            w.close()
        old_chunk = voice._DELTA_CHUNK_ELEMS
        voice._DELTA_CHUNK_ELEMS = 100  # small and unaligned vs Q8_0 block 32
        max_slice = []
        orig_slice = voice._gguf_raw_slice_to_f32
        def _spy(raw, qt, off, n):
            max_slice.append(n)
            return orig_slice(raw, qt, off, n)
        voice._gguf_raw_slice_to_f32 = _spy
        try:
            out = voice._compute_delta(str(vp), str(bp), d / "out", "delta-x", ns())
        finally:
            voice._DELTA_CHUNK_ELEMS = old_chunk
            voice._gguf_raw_slice_to_f32 = orig_slice
        self.assertGreater(len(max_slice), 2, "expected many bounded chunks")
        self.assertLessEqual(max(max_slice), 100, max_slice[:5])
        vsrc, bsrc = voice.open_source(str(vp)), voice.open_source(str(bp))
        want = (vsrc.read_f32("output.weight").astype(np.float32)
                - bsrc.read_f32("output.weight").astype(np.float32)).astype("float16")
        got = read_arr(out, "output.weight")
        self.assertTrue((got == want).all())
    def test_share_safe_delta_header(self):
        import json as _j
        d = _mktmp(self)
        home = d / "Users" / "someone" / "Desktop"
        home.mkdir(parents=True)
        self._pair(home)
        out = voice._compute_delta(str(home / "v.safetensors"),
                                   str(home / "b.safetensors"),
                                   d / "out", "delta-x", ns())
        outp = Path(str(out))
        data = (outp / "voice.safetensors").read_bytes() \
            if outp.is_dir() else outp.read_bytes()
        hs = struct.unpack("<Q", data[:8])[0]
        meta = _j.loads(data[8:8 + hs]).get("__metadata__", {})
        blob = _j.dumps(meta)
        self.assertNotIn("Users", blob)
        self.assertNotIn("someone", blob)
        self.assertEqual(meta.get("base"), "b.safetensors")
        self.assertEqual(meta.get("voice"), "v.safetensors")

    def test_verify_agrees(self):
        d = _mktmp(self)
        write_st(d / "v.safetensors",
                 {"output.weight": (np.zeros((4, 4), dtype=np.float16), "F16")})
        meta = {"tensor_name": "output.weight", "shape": [4, 4],
                "dtype": "F16", "type": "voice"}
        self.assertEqual(voice._verify_voice_file(str(d / "v.safetensors"), meta), [])

    def test_verify_flags_mismatch(self):
        d = _mktmp(self)
        write_st(d / "v.safetensors",
                 {"output.weight": (np.zeros((4, 4), dtype=np.float16), "F16")})
        meta = {"tensor_name": "output.weight", "shape": [262144, 2048],
                "dtype": "BF16", "type": "voice"}
        warns = voice._verify_voice_file(str(d / "v.safetensors"), meta)
        blob = "\n".join(warns)
        self.assertIn("MISMATCH shape", blob)
        self.assertIn("MISMATCH dtype", blob)
        self.assertIn("file wins", blob)

    def test_gguf_slice_q4k_superblocks(self):
        # K-quants share scales across 256-elem superblocks: slices crossing
        # superblock boundaries must equal full dequantize cut to the same range.
        # (gguf lib can't quantize Q4_K here, so synthesize raw superblock bytes.)
        from gguf.constants import GGMLQuantizationType as Q
        from gguf.quants import dequantize
        import numpy as _np
        elems, bs, ts = 1024, 256, 144
        raw = _np.random.default_rng(3).integers(0, 256, size=elems // bs * ts).astype(_np.uint8).tobytes()
        full = dequantize(_np.frombuffer(raw, dtype=_np.uint8).reshape(-1, ts), Q.Q4_K).reshape(-1)
        for off, n in ((0, 100), (200, 100), (100, 500), (900, 124), (0, 1024)):
            b0 = off // bs
            b1 = (off + n + bs - 1) // bs
            seg = raw[b0 * ts:b1 * ts]
            got = voice._gguf_raw_slice_to_f32(seg, Q.Q4_K, off - b0 * bs, n)
            self.assertTrue((got == full[off:off + n]).all(), f"off={off} n={n}")

    def test_niche_gates_fail_fast_non_interactive(self):
        for mv, mb in ((False, True), (True, False), (True, True)):
            d = _mktmp(self)
            v, b = self._pair(d, marker_voice=mv, marker_base=mb)
            with self.assertRaises(SystemExit, msg=f"mv={mv} mb={mb}"):
                voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())


class TestGetGguf(IsolatedVoices):
    def test_get_from_local_gguf(self):
        from gguf import GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        gp = d / "tiny.gguf"
        w = GGUFWriter(str(gp), "llama", use_temp_file=False)
        w.add_architecture()
        w.add_tensor("output.weight",
                     np.zeros((16, 16), dtype=np.float16), raw_dtype=Q.F16)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        voice.cmd_get(ns(source=str(gp), name="tiny-gguf"))
        self.assertTrue((self.vdir / "tiny-gguf" / "voice.safetensors").exists())

    def test_get_from_split_repo_picks_head(self):
        import types
        from unittest import mock
        from vlib import cmds_voices
        names = ["blk.0.ffn_down.weight", "output.weight"]
        stub = types.SimpleNamespace(
            kind="gguf", path=None, repo=None, _reader=None, _remote=None,
            names=lambda: names,
            ref=lambda n: voice.TensorRef(n, "BF16", (8, 4)),
            _load=lambda: None)
        arr = np.zeros((8, 4), dtype=np.float16)
        with mock.patch.object(cmds_voices, "open_source", return_value=stub), \
             mock.patch.object(cmds_voices, "_fetch_tensor",
                               return_value=("bytes", None, arr.tobytes(), "BF16", (8, 4))) as ft:
            voice.cmd_get(ns(source="org/m", name="x"))
        self.assertEqual(ft.call_args[0][1], "output.weight")
        self.assertTrue((self.vdir / "x" / "voice.safetensors").exists())

class TestGgufTensorFirst(unittest.TestCase):
    def _parsed(self, tensors):
        from gguf.constants import GGMLQuantizationType as Q
        off, table = 0, {}
        for n, (dims, dt) in tensors.items():
            table[n] = (list(dims), int(Q[dt]), off, 8)
            off += 8
        return ("llama", 64, table)

    def _patch_parse(self, mapping):
        from unittest import mock
        import vlib.sources as _src
        return mock.patch.object(_src, "_parse_gguf_header", side_effect=lambda u: mapping[u])

    def test_triage_groups_splits_drops_polluters(self):
        files = ["BF16/m-00002-of-00002.gguf", "BF16/m-00001-of-00002.gguf",
                 "m-Q8_0.gguf", "mmproj-m.gguf", "x-imatrix.gguf"]
        g, s, d = voice._triage_gguf_files(files)
        self.assertEqual(g, [["BF16/m-00001-of-00002.gguf", "BF16/m-00002-of-00002.gguf"]])
        self.assertEqual(s, ["m-Q8_0.gguf"])
        self.assertEqual(sorted(d), ["mmproj-m.gguf", "x-imatrix.gguf"])

    def test_triage_singleton_split_counts_as_single(self):
        g, s, _d = voice._triage_gguf_files(["m-00001-of-00001.gguf"])
        self.assertEqual(g, [])
        self.assertEqual(s, ["m-00001-of-00001.gguf"])

    def test_census_requires_head_and_body(self):
        self.assertEqual(voice._gguf_unit_census(["blk.0.attn.weight", "output.weight"]), ("output.weight", True))
        self.assertEqual(voice._gguf_unit_census(["blk.0.attn.weight", "token_embd.weight"]), ("token_embd.weight", True))
        self.assertIsNone(voice._gguf_unit_census(["blk.0.attn.weight"])[0])
        self.assertFalse(voice._gguf_unit_census(["output.weight"])[1])
        self.assertEqual(voice._gguf_unit_census(["proj.weight"] * 1), (None, False))

    def test_dtype_rank_orders_fidelity(self):
        r = voice._gguf_dtype_rank
        self.assertTrue(r("F32") > r("BF16") > r("F16") > r("Q8_0") > r("Q4_K") > r("IQ1_M") > r("NVFP4"))
        self.assertEqual(r("NOPE"), -1)

    def test_multi_source_union_prefers_best_fidelity(self):
        u1 = "https://h/x/a.gguf"
        u2 = "https://h/x/b.gguf"
        mapping = {
            u1: self._parsed({"blk.0.attn.weight": ([4, 4], "Q8_0"), "output.weight": ([8, 4], "Q8_0")}),
            u2: self._parsed({"blk.1.attn.weight": ([4, 4], "BF16"), "output.weight": ([8, 4], "BF16")}),
        }
        with self._patch_parse(mapping):
            src = voice.GgufSource(urls=[u1, u2])
            self.assertEqual(src.names(), ["blk.0.attn.weight", "output.weight", "blk.1.attn.weight"])
            self.assertEqual(src.ref("output.weight").dtype, "BF16")
            self.assertEqual(src.ref("output.weight").shape, (4, 8))
            with self.assertRaises(KeyError):
                src.ref("ghost.weight")

    def test_select_split_repo_for_head_shard(self):
        b = "https://huggingface.co/o/m/resolve/main/"
        files = ["m-BF16-00001-of-00002.gguf", "m-BF16-00002-of-00002.gguf", "m-Q8_0.gguf"]
        mapping = {
            b + "m-BF16-00001-of-00002.gguf": self._parsed({"blk.0.ffn_down.weight": ([8, 4], "BF16")}),
            b + "m-BF16-00002-of-00002.gguf": self._parsed({"output.weight": ([8, 4], "BF16")}),
            b + "m-Q8_0.gguf": self._parsed({"blk.0.attn.weight": ([4, 4], "Q8_0"), "output.weight": ([8, 4], "Q8_0")}),
        }
        with self._patch_parse(mapping):
            src = voice._gguf_source_for_repo("o/m", "main", files)
            self.assertEqual(src.urls, [b + "m-BF16-00001-of-00002.gguf", b + "m-BF16-00002-of-00002.gguf"])
            self.assertIn("output.weight", src.names())

    def test_select_prefers_single_when_split_headless(self):
        b = "https://huggingface.co/o/m/resolve/main/"
        files = ["m-00001-of-00002.gguf", "m-00002-of-00002.gguf", "m-Q8_0.gguf"]
        mapping = {
            b + "m-00001-of-00002.gguf": self._parsed({"blk.0.attn.weight": ([4, 4], "F16")}),
            b + "m-00002-of-00002.gguf": self._parsed({"blk.1.attn.weight": ([4, 4], "F16")}),
            b + "m-Q8_0.gguf": self._parsed({"blk.0.attn.weight": ([4, 4], "Q8_0"), "output.weight": ([8, 4], "Q8_0")}),
        }
        with self._patch_parse(mapping):
            src = voice._gguf_source_for_repo("o/m", "main", files)
            self.assertEqual(src.url, b + "m-Q8_0.gguf")

    def test_select_fails_loud_when_nothing_resolves(self):
        b = "https://huggingface.co/o/m/resolve/main/"
        with self._patch_parse({b + "mmproj-x.gguf": self._parsed({"proj.weight": ([4, 4], "F16")})}):
            with self.assertRaises(SystemExit):
                voice._gguf_source_for_repo("o/m", "main", ["mmproj-x.gguf"])

    def test_select_finds_valid_unit_past_old_cap(self):
        b = "https://huggingface.co/o/m/resolve/main/"
        files = [f"a{i}-Q8_0.gguf" for i in range(8)] + ["z-model.gguf"]
        mapping = {b + f: self._parsed({"proj.weight": ([4, 4], "F16")}) for f in files[:8]}
        mapping[b + "z-model.gguf"] = self._parsed({"blk.0.attn.weight": ([4, 4], "F16"),
                                                   "output.weight": ([8, 4], "F16")})
        with self._patch_parse(mapping):
            src = voice._gguf_source_for_repo("o/m", "main", files)
            self.assertEqual(src.url, b + "z-model.gguf")

class TestFeedback(unittest.TestCase):
    def test_fetch_bytes_small_uses_single_read(self):
        from unittest import mock
        import vlib.net as net
        with mock.patch.object(net, "http_range", return_value=(b"xy", 206, None)) as hr, \
             mock.patch.object(net, "download_range") as dr:
            self.assertEqual(net._fetch_bytes_progress("http://x/f", 0, 1, "t"), b"xy")
            dr.assert_not_called()
            hr.assert_called_once()

    def test_fetch_bytes_large_progresses_and_cleans(self):
        from unittest import mock
        import vlib.net as net
        from vlib import ctx
        d = _mktmp(self)
        big = net._PROGRESS_BYTES + 8
        def fake_dl(url, s, e, part_dir, etag, label=""):
            part_dir.mkdir(parents=True, exist_ok=True)
            (part_dir / "combined.part").write_bytes(b"z" * (e - s + 1))
            return part_dir / "combined.part"
        with mock.patch.object(ctx, "VOICES_DIR", d), \
             mock.patch.object(net, "download_range", side_effect=fake_dl) as dr, \
             mock.patch.object(net, "http_range") as hr:
            self.assertEqual(net._fetch_bytes_progress("http://x/f.gguf", 0, big - 1, "head"), b"z" * big)
            hr.assert_not_called()
            self.assertEqual(dr.call_args[1].get("label"), "head")
            self.assertFalse(any(p.is_file() for p in d.rglob("*")), "part files cleaned on success")

    def test_fetch_bytes_failure_keeps_parts(self):
        from unittest import mock
        import vlib.net as net
        from vlib import ctx
        d = _mktmp(self)
        def fake_dl(url, s, e, part_dir, etag, label=""):
            part_dir.mkdir(parents=True, exist_ok=True)
            (part_dir / "chunk_000.part").write_bytes(b"z")
            raise RuntimeError("boom")
        with mock.patch.object(ctx, "VOICES_DIR", d), \
             mock.patch.object(net, "download_range", side_effect=fake_dl):
            with self.assertRaises(RuntimeError):
                net._fetch_bytes_progress("http://x/f.gguf", 0, net._PROGRESS_BYTES + 1, "head")
            self.assertTrue(any(p.name == "chunk_000.part" for p in (d / ".parts").rglob("*")),
                            "parts stay on failure so the next run resumes")

    def test_tied_note_respects_quiet(self):
        import io
        from contextlib import redirect_stdout
        from unittest import mock
        from vlib import ctx
        d = _mktmp(self)
        v, b = d / "v.safetensors", d / "b.safetensors"
        write_st(v, {"lm_head.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        write_st(b, {"model.language_model.embed_tokens.weight": (np.full((4, 4), 1.0, dtype=np.float16), "F16")})
        buf = io.StringIO()
        with mock.patch.object(ctx, "QUIET", True), redirect_stdout(buf):
            voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())
        self.assertNotIn("ties its head", buf.getvalue(), "filler must not leak through -q")

    def test_lookup_opener_and_quiet(self):
        import io
        from contextlib import redirect_stdout
        from unittest import mock
        from vlib import cmds_voices, ctx
        with mock.patch.object(cmds_voices, "open_source", side_effect=RuntimeError("net down")):
            buf = io.StringIO()
            with redirect_stdout(buf):
                with self.assertRaises(SystemExit):
                    voice.cmd_get(ns(source="org/model", name="x"))
            self.assertIn("Looking up org/model", buf.getvalue())
            buf2 = io.StringIO()
            with mock.patch.object(ctx, "QUIET", True), redirect_stdout(buf2):
                with self.assertRaises(SystemExit):
                    voice.cmd_get(ns(source="org/model", name="x"))
            self.assertNotIn("Looking up", buf2.getvalue())

    def test_using_opener_and_quiet(self):
        import io
        from contextlib import redirect_stdout
        from unittest import mock
        from vlib import ctx
        b = "https://huggingface.co/o/m/resolve/main/"
        mapping = {b + "m-Q8_0.gguf": (("llama", 64, {"blk.0.attn.weight": ([4, 4], 8, 0, 8),
                                                      "output.weight": ([8, 4], 8, 0, 8)}))}
        with mock.patch("vlib.sources._parse_gguf_header", side_effect=lambda u: mapping[u]):
            buf = io.StringIO()
            with redirect_stdout(buf):
                voice._gguf_source_for_repo("o/m", "main", ["m-Q8_0.gguf"])
            self.assertIn("Using m-Q8_0.gguf", buf.getvalue())
            buf2 = io.StringIO()
            with mock.patch.object(ctx, "QUIET", True), redirect_stdout(buf2):
                voice._gguf_source_for_repo("o/m", "main", ["m-Q8_0.gguf"])
            self.assertNotIn("Using", buf2.getvalue())

    def test_saving_opener_and_quiet(self):
        import io
        from contextlib import redirect_stdout
        from unittest import mock
        from vlib import ctx
        d = _mktmp(self)
        arr = np.full((4, 4), 1.0, dtype=np.float16)
        fetched = ("bytes", None, arr.tobytes(), "F16", (4, 4))
        buf = io.StringIO()
        with redirect_stdout(buf):
            voice._write_voice(d / "v", "output.weight", fetched, {"name": "x"})
        self.assertIn("Saving voice", buf.getvalue())
        self.assertTrue((d / "v" / "voice.safetensors").exists())
        buf2 = io.StringIO()
        with mock.patch.object(ctx, "QUIET", True), redirect_stdout(buf2):
            voice._write_voice(d / "v2", "output.weight", fetched, {"name": "x"})
        self.assertNotIn("Saving", buf2.getvalue())

    def test_preparing_opener_and_quiet(self):
        import io
        from contextlib import redirect_stdout
        from unittest import mock
        from vlib import cmds_voices, ctx
        d = _mktmp(self)
        v = d / "v.safetensors"
        write_st(v, {"output.weight": (np.full((4, 4), 2.0, dtype=np.float16), "F16")})
        with mock.patch.object(cmds_voices, "_ensure_base_tensors", side_effect=SystemExit(1)):
            buf = io.StringIO()
            with redirect_stdout(buf):
                with self.assertRaises(SystemExit):
                    voice.cmd_delta(ns(voicename=str(v), base="org/model"))
            self.assertIn("Preparing base", buf.getvalue())
            buf2 = io.StringIO()
            with mock.patch.object(ctx, "QUIET", True), redirect_stdout(buf2):
                with self.assertRaises(SystemExit):
                    voice.cmd_delta(ns(voicename=str(v), base="org/model"))
            self.assertNotIn("Preparing", buf2.getvalue())

    def test_source_read_f32_large_uses_progress_path(self):
        import struct
        from unittest import mock
        import vlib.net as net
        from vlib import ctx
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        raw64 = struct.pack("<16f", *([1.0] * 16))
        table = {"output.weight": ([4, 4], int(Q.F32), 0, 20000000)}
        parsed = ("llama", 64, table)
        def fake_dl(url, s, e, part_dir, etag, label=""):
            part_dir.mkdir(parents=True, exist_ok=True)
            (part_dir / "combined.part").write_bytes(raw64)
            return part_dir / "combined.part"
        with mock.patch("vlib.sources._parse_gguf_header", return_value=parsed), \
             mock.patch.object(ctx, "VOICES_DIR", d), \
             mock.patch.object(net, "download_range", side_effect=fake_dl) as dr:
            src = voice.GgufSource(urls=["http://x/m.gguf"])
            got = src.read_f32("output.weight")
            self.assertEqual(tuple(got.shape), (4, 4))
            self.assertTrue((got == 1.0).all())
            self.assertTrue(dr.called, "large tensor must go through the progress path")


class TestQuantShapes(unittest.TestCase):
    def _q8_voice(self, d, shape=(64, 96)):
        from gguf import GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        from gguf.quants import quantize
        fused = np.arange(int(np.prod(shape)), dtype=np.float32).reshape(shape)
        q = quantize(fused, Q.Q8_0)
        vp = d / "v.gguf"
        w = GGUFWriter(str(vp), "llama", use_temp_file=False)
        w.add_tensor("output.weight", q, raw_dtype=Q.Q8_0)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        return vp, fused

    def test_single_tensor_artifact_keeps_logical_shape(self):
        from gguf.constants import GGMLQuantizationType as Q
        from gguf.quants import quantize
        d = _mktmp(self)
        vp, fused = self._q8_voice(d)
        src = voice.open_source(str(vp))
        ref = src.ref("output.weight")
        raw = src.read_raw("output.weight")
        out = str(d / "voice.gguf")
        Path(d / "raw.bin").write_bytes(raw)
        voice._write_single_tensor_gguf("output.weight", d / "raw.bin", len(raw),
                                        ref.dtype, ref.shape, out)
        back = voice.open_source(out)
        self.assertEqual(tuple(back.ref("output.weight").shape), tuple(ref.shape))
        self.assertEqual(back.read_raw("output.weight"), raw)

    def test_transplant_cast_keeps_shape_and_bytes(self):
        from gguf import GGUFReader, GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        vp, _ = self._q8_voice(d)
        tp = d / "t.gguf"
        w = GGUFWriter(str(tp), "llama", use_temp_file=False)
        w.add_architecture()
        w.add_tensor("output.weight", np.zeros((64, 96), dtype=np.float16),
                     raw_dtype=Q.F16)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        out = d / "voiced.gguf"
        voice.cmd_cast(ns(target=str(tp), voice=str(vp), out=str(out)))
        got = [t for t in GGUFReader(str(out)).tensors if t.name == "output.weight"][0]
        self.assertEqual(tuple(reversed(tuple(int(x) for x in got.shape))), (64, 96))
        src_raw = voice.open_source(str(vp)).read_raw("output.weight")
        self.assertEqual(bytes(got.data), src_raw)



class TestVoicePacks(IsolatedVoices):
    def _pack_pair(self, d):
        write_st(d / "v.safetensors", {
            "output.weight": (np.full((4, 4), 3.0, np.float16), "F16"),
            "model.norm.weight": (np.full((8,), 5.0, np.float16), "F16")})
        write_st(d / "b.safetensors", {
            "output.weight": (np.full((4, 4), 2.0, np.float16), "F16"),
            "model.norm.weight": (np.full((8,), 4.0, np.float16), "F16"),
            "extra.weight": (np.zeros((2, 2), np.float16), "F16")})
        return d / "v.safetensors", d / "b.safetensors"

    def test_pack_delta_computes_all(self):
        d = _mktmp(self)
        v, b = self._pack_pair(d)
        out = voice._compute_delta(str(v), str(b), d / "out", "delta-x", ns())
        for name, want in (("output.weight", 1.0), ("model.norm.weight", 1.0)):
            got = read_arr(out, name)
            self.assertTrue((got == want).all(), name)

    def test_pack_delta_missing_name_fails(self):
        d = _mktmp(self)
        write_st(d / "v.safetensors",
                 {"ghost.weight": (np.zeros((2, 2), np.float16), "F16")})
        write_st(d / "b.safetensors",
                 {"other.weight": (np.zeros((2, 2), np.float16), "F16")})
        with self.assertRaises(SystemExit):
            voice._compute_delta(str(d / "v.safetensors"), str(d / "b.safetensors"),
                                 d / "out", "delta-x", ns())

    def test_tensor_cast_delta_fuses_not_replaces(self):
        from gguf import GGUFReader, GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        write_st(d / "v.safetensors",
                 {"output.weight": (np.full((4, 4), 3.0, np.float16), "F16")})
        write_st(d / "b.safetensors",
                 {"output.weight": (np.full((4, 4), 2.0, np.float16), "F16")})
        dout = voice._compute_delta(str(d / "v.safetensors"), str(d / "b.safetensors"),
                                    d / "dout", "delta-x", ns())
        tp = d / "t.gguf"
        w = GGUFWriter(str(tp), "llama", use_temp_file=False)
        w.add_architecture()
        w.add_tensor("output.weight", np.full((4, 4), 10.0, np.float16), raw_dtype=Q.F16)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        out = d / "voiced.gguf"
        voice.cmd_cast(ns(target=str(tp), voice=str(dout), out=str(out)))
        got = [t for t in GGUFReader(str(out)).tensors if t.name == "output.weight"][0]
        # base 10 + delta (3-2) = 11. Raw-delta replace would give 1.
        self.assertTrue((np.asarray(got.data, dtype=np.float32) == 11.0).all())

    def test_pack_cast_gguf_fuses_all(self):
        from gguf import GGUFReader, GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        v, b = self._pack_pair(d)
        dout = voice._compute_delta(str(v), str(b), d / "dout", "delta-x", ns())
        tp = d / "t.gguf"
        w = GGUFWriter(str(tp), "llama", use_temp_file=False)
        w.add_architecture()
        w.add_tensor("output.weight", np.full((4, 4), 10.0, np.float16), raw_dtype=Q.F16)
        w.add_tensor("model.norm.weight", np.full((8,), 20.0, np.float32), raw_dtype=Q.F32)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        out = d / "voiced.gguf"
        voice.cmd_cast(ns(target=str(tp), voice=str(dout), out=str(out)))
        back = {t.name: t for t in GGUFReader(str(out)).tensors}
        self.assertTrue((np.asarray(back["output.weight"].data, dtype=np.float32) == 11.0).all())
        self.assertTrue((np.asarray(back["model.norm.weight"].data, dtype=np.float32) == 21.0).all())

    def test_pack_cast_safetensors_fuses_all(self):

        d = _mktmp(self)
        v, b = self._pack_pair(d)
        dout = voice._compute_delta(str(v), str(b), d / "dout", "delta-x", ns())
        write_st(d / "t.safetensors", {
            "output.weight": (np.full((4, 4), 10.0, np.float16), "F16"),
            "model.norm.weight": (np.full((8,), 20.0, np.float16), "F16")})
        out = d / "o.safetensors"
        voice.cmd_cast(ns(target=str(d / "t.safetensors"),
                          voice=str(dout), out=str(out)))
        for name, want in (("output.weight", 11.0), ("model.norm.weight", 21.0)):
            got = read_arr(out, name)
            self.assertTrue((got == want).all(), name)

    def test_pack_cast_safetensors_single_pass(self):
        # The old chain called rewrite_safetensors_tensor once per tensor
        # (O(N.file) full copies). The fused path must succeed with it disabled.
        d = _mktmp(self)
        v, b = self._pack_pair(d)
        dout = voice._compute_delta(str(v), str(b), d / "dout", "delta-x", ns())
        write_st(d / "t.safetensors", {
            "output.weight": (np.full((4, 4), 10.0, np.float16), "F16"),
            "model.norm.weight": (np.full((8,), 20.0, np.float16), "F16")})
        out = d / "o.safetensors"
        orig = voice.rewrite_safetensors_tensor
        def _boom(*a, **k):
            raise AssertionError("chained per-tensor rewrite used")
        voice.rewrite_safetensors_tensor = _boom
        try:
            voice.cmd_cast(ns(target=str(d / "t.safetensors"),
                              voice=str(dout), out=str(out)))
        finally:
            voice.rewrite_safetensors_tensor = orig
        for name, want in (("output.weight", 11.0), ("model.norm.weight", 21.0)):
            got = read_arr(out, name)
            self.assertTrue((got == want).all(), name)


class TestBaseCache(unittest.TestCase):
    def _cache(self, d, tensors):
        write_st(d / "base.safetensors", tensors)
        return d / "base.safetensors"

    def test_validate_hit(self):
        d = _mktmp(self)
        p = self._cache(d, {"a": (np.zeros((4, 4), np.float16), "F16")})
        self.assertEqual(voice._cache_validate(str(p), ["a"]), [])

    def test_validate_missing_and_truncated(self):
        import struct as _st, json as _j
        d = _mktmp(self)
        p = self._cache(d, {"a": (np.zeros((4, 4), np.float16), "F16")})
        raw = bytearray(p.read_bytes())
        hs = _st.unpack("<Q", bytes(raw[:8]))[0]
        hdr = _j.loads(bytes(raw[8:8 + hs]))
        hdr["b"] = {"dtype": "F16", "shape": [8], "data_offsets": [32, 48]}
        hj = _j.dumps(hdr).encode()
        p.write_bytes(_st.pack("<Q", len(hj)) + hj + bytes(raw[8 + hs:]))
        self.assertEqual(sorted(voice._cache_validate(str(p), ["a", "b", "c"])), ["b", "c"])

    def test_validate_garbage(self):
        d = _mktmp(self)
        p = d / "base.safetensors"
        p.write_bytes(b"not-a-safetensors-file")
        self.assertEqual(voice._cache_validate(str(p), ["a"]), ["a"])

    def test_topup_write_roundtrip(self):
        d = _mktmp(self)
        p = self._cache(d, {"a": (np.full((4, 4), 7.0, np.float16), "F16")})
        rawf = d / "raw.bin"
        rawf.write_bytes(np.full((8,), 9.0, np.float16).tobytes())
        staged = [("b", np.full((2, 2), 3.0, np.float16).tobytes(), "F16", (2, 2), None),
                  ("c", None, "F16", (8,), rawf)]
        new_hdr = voice._cache_topup_write(p, staged)
        self.assertEqual(sorted(k for k in new_hdr if k != "__metadata__"), ["a", "b", "c"])
        for name, want in (("a", 7.0), ("b", 3.0), ("c", 9.0)):
            self.assertTrue((read_arr(p, name) == want).all(), name)


class TestRunGuarded(unittest.TestCase):
    def test_timeout_reaps_child(self):
        import subprocess
        import time
        from _helpers import run_guarded
        child = []
        orig_popen = subprocess.Popen

        class Spy(orig_popen):
            def __init__(self, *a, **k):
                super().__init__(*a, **k)
                child.append(self)

        subprocess.Popen = Spy
        try:
            t0 = time.time()
            with self.assertRaises(subprocess.TimeoutExpired):
                run_guarded(["sleep", "30"], timeout=2)
            dt = time.time() - t0
        finally:
            subprocess.Popen = orig_popen
        self.assertLess(dt, 15, dt)
        self.assertIsNotNone(child[0].poll(), "child left running")

    def test_fast_command_passes_through(self):
        from _helpers import run_guarded
        r = run_guarded(["echo", "hi"], timeout=30)
        self.assertEqual((r.returncode, r.stdout.strip()), (0, "hi"))

class TestAliases(unittest.TestCase):
    GGUF_NAMES = ["token_embd.weight", "blk.28.attn_o.weight", "blk.29.attn_o.weight",
                  "blk.29.ffn_down.weight", "output_norm.weight", "output.weight"]

    def test_alias_both_directions(self):
        self.assertEqual(voice._pack_aliases("model.language_model.embed_tokens.weight", self.GGUF_NAMES),
                         ["token_embd.weight"])
        self.assertEqual(voice._pack_aliases("token_embd.weight",
                         ["model.language_model.embed_tokens.weight", "output.weight"]),
                         ["model.language_model.embed_tokens.weight"])
        self.assertEqual(voice._pack_aliases("model.language_model.layers.29.self_attn.o_proj.weight",
                                             self.GGUF_NAMES),
                         ["blk.29.attn_o.weight"])
        self.assertEqual(voice._pack_aliases("lm_head.weight", self.GGUF_NAMES), ["output.weight"])

    def test_alias_block_mismatch_is_no_match(self):
        self.assertEqual(voice._pack_aliases("model.language_model.layers.28.self_attn.o_proj.weight",
                                             ["blk.29.attn_o.weight"]), [])
        self.assertEqual(voice._pack_aliases("someday.weight", self.GGUF_NAMES), [])

    def test_alias_ambiguous_lists_all(self):
        self.assertEqual(sorted(voice._pack_aliases("model.embed_tokens.weight",
                                                    ["a.token_embd.weight", "b.token_embd.weight"])),
                         ["a.token_embd.weight", "b.token_embd.weight"])

    def test_alias_modern_gguf_variants(self):
        # Evidenced live: Gemma3/Qwen3 GGUFs use attn_output + post_attention_norm.
        names = ["blk.29.attn_output.weight", "blk.29.post_attention_norm.weight"]
        self.assertEqual(voice._pack_aliases("model.layers.29.self_attn.o_proj.weight", names),
                         ["blk.29.attn_output.weight"])
        self.assertEqual(voice._pack_aliases("model.layers.29.post_attention_layernorm.weight", names),
                         ["blk.29.post_attention_norm.weight"])

    def test_alias_glm_generations(self):
        # Evidenced live: ChatGLM-style HF names vs llama.cpp GGUF names.
        gguf = ["token_embd.weight", "blk.39.attn_output.weight",
                "blk.39.ffn_down.weight", "output_norm.weight", "output.weight"]
        self.assertEqual(voice._pack_aliases("transformer.encoder.layers.39.self_attention.dense.weight", gguf),
                         ["blk.39.attn_output.weight"])
        self.assertEqual(voice._pack_aliases("transformer.encoder.layers.39.mlp.dense_4h_to_h.weight", gguf),
                         ["blk.39.ffn_down.weight"])
        self.assertEqual(voice._pack_aliases("transformer.encoder.final_layernorm.weight", gguf),
                         ["output_norm.weight"])
        self.assertEqual(voice._pack_aliases("transformer.output_layer.weight", gguf),
                         ["output.weight"])

    def test_pack_preset_head_only(self):
        # Preset is the output head, nothing else. Tied repos resolve the embed
        # (it doubles as head); untied resolve lm_head.
        d = _mktmp(self)
        write_st(d / "tied.safetensors", {
            "model.language_model.embed_tokens.weight": (np.zeros((8, 4), np.float16), "F16"),
            "model.language_model.norm.weight": (np.zeros((4,), np.float16), "F16")})
        write_st(d / "untied.safetensors", {
            "lm_head.weight": (np.zeros((8, 4), np.float16), "F16"),
            "model.embed_tokens.weight": (np.zeros((8, 4), np.float16), "F16"),
            "model.norm.weight": (np.zeros((4,), np.float16), "F16")})
        src = voice.open_source(str(d / "tied.safetensors"))
        self.assertEqual(_resolve_pack_tensors(src, "r", None, src.names()),
                         ["model.language_model.embed_tokens.weight"])
        src = voice.open_source(str(d / "untied.safetensors"))
        self.assertEqual(_resolve_pack_tensors(src, "r", None, src.names()),
                         ["lm_head.weight"])

    def test_alias_moe_experts(self):
        # Evidenced live: Qwen3-30B-A3B GGUF experts vs HF experts.* names.
        gguf = ["blk.5.ffn_down_exps.weight", "blk.5.ffn_gate_inp.weight"]
        self.assertEqual(voice._pack_aliases("model.layers.5.experts.down_proj", gguf),
                         ["blk.5.ffn_down_exps.weight"])
        # fused HF gate_up has no 1:1 GGUF counterpart (split exps) — no match, fail loud downstream
        self.assertEqual(voice._pack_aliases("model.layers.5.experts.gate_up_proj", gguf), [])
        # routers never map (semantics differ per arch)
        self.assertEqual(voice._pack_aliases("model.layers.5.router.per_expert_scale", gguf), [])

    def test_pack_cast_bridges_naming(self):
        # ST pack names onto a GGUF target speaking llama.cpp names (replace mode).
        from gguf import GGUFReader, GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        d = _mktmp(self)
        write_st(d / "pack.safetensors", {
            "model.language_model.embed_tokens.weight": (np.full((4, 4), 3.0, np.float16), "F16"),
            "lm_head.weight": (np.full((4, 4), 7.0, np.float16), "F16")})
        tp = d / "t.gguf"
        w = GGUFWriter(str(tp), "llama", use_temp_file=False)
        w.add_architecture()
        w.add_tensor("token_embd.weight", np.zeros((4, 4), dtype=np.float16), raw_dtype=Q.F16)
        w.add_tensor("output.weight", np.zeros((4, 4), dtype=np.float16), raw_dtype=Q.F16)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        out = d / "voiced.gguf"
        voice.cmd_cast(ns(target=str(tp), voice=str(d / "pack.safetensors"), out=str(out)))
        back = {t.name: np.asarray(t.data, dtype=np.float32) for t in GGUFReader(str(out)).tensors}
        self.assertTrue((back["token_embd.weight"] == 3.0).all())
        self.assertTrue((back["output.weight"] == 7.0).all())


    def test_nearest_encodable_mapping(self):
        from gguf.constants import GGMLQuantizationType as Q
        self.assertEqual(voice._nearest_encodable(Q.IQ4_NL), Q.Q4_0)  # same 4.5bpp
        self.assertEqual(voice._nearest_encodable(Q.Q4_K), Q.Q4_0)
        self.assertEqual(voice._nearest_encodable(Q.Q6_K), Q.Q5_1)

    def test_unencodable_target_keeps_nearest(self):
        # Q4_K target (decode-only in gguf lib): fused output lands on Q4_0, not F16.
        from gguf import GGUFReader, GGUFWriter
        from gguf.constants import GGMLQuantizationType as Q
        from gguf.quants import quant_shape_to_byte_shape
        d = _mktmp(self)
        logical = (256, 256)
        raw = np.zeros(quant_shape_to_byte_shape(logical, Q.Q4_K), dtype=np.uint8)
        tp = d / "t.gguf"
        w = GGUFWriter(str(tp), "llama", use_temp_file=False)
        w.add_architecture()
        w.add_tensor("blk.0.ffn_down.weight", raw, raw_dtype=Q.Q4_K)
        w.write_header_to_file()
        w.write_kv_data_to_file()
        w.write_tensors_to_file()
        w.close()
        got = [t for t in GGUFReader(str(tp)).tensors if t.name == "blk.0.ffn_down.weight"][0]
        self.assertEqual(got.tensor_type, Q.Q4_K)
        write_st(d / "v.safetensors",
                 {"blk.0.ffn_down.weight": (np.full(logical, 1.0, np.float16), "F16")})
        write_st(d / "b.safetensors",
                 {"blk.0.ffn_down.weight": (np.zeros(logical, np.float16), "F16")})
        dout = voice._compute_delta(str(d / "v.safetensors"), str(d / "b.safetensors"),
                                    d / "dout", "delta-x", ns())
        out = d / "voiced.gguf"
        voice.cmd_cast(ns(target=str(tp), voice=str(dout), out=str(out)))
        fused = [t for t in GGUFReader(str(out)).tensors if t.name == "blk.0.ffn_down.weight"][0]
        self.assertEqual(fused.tensor_type, Q.Q4_0)


    def test_batch_miss_reports_all_with_hints(self):
        # One failure names EVERY miss (no fix-forward) with same-size hints.
        import io
        from contextlib import redirect_stdout
        names = ["token_embd.weight", "blk.29.attn_o.weight", "other.weight"]
        counts = {"token_embd.weight": 16, "blk.29.attn_o.weight": 16, "other.weight": 64}
        pairs = [("model.language_model.layers.29.self_attn.o_proj.weight", (4, 4)),
                 ("ghost.weight", (8, 8)),
                 ("phantom.weight", (2, 2))]
        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit):
                voice._resolve_pack_targets(pairs, names, "t.gguf", counts)
        blob = buf.getvalue()
        self.assertIn("ghost.weight", blob)
        self.assertIn("phantom.weight", blob)
        self.assertIn("other.weight", blob)  # same-size hint for the 8x8 ghost
        self.assertIn("2 tensor(s)", blob)  # both misses at once; the aliased one is not listed
        # all-resolvable returns the mapping (exact + alias)
        got = voice._resolve_pack_targets(
            [("model.language_model.layers.29.self_attn.o_proj.weight", (4, 4)),
             ("token_embd.weight", (4, 4))],
            names, "t.gguf", counts)
        self.assertEqual(got, {"model.language_model.layers.29.self_attn.o_proj.weight": "blk.29.attn_o.weight",
                               "token_embd.weight": "token_embd.weight"})

    def test_pack_cast_bridges_same_format_prefix(self):
        # Qwen-style prefixed voice onto bare names: same (block, trailing
        # role) + equal count maps without growing _ROLE_ALIASES.
        import io
        from contextlib import redirect_stdout
        got = voice._resolve_pack_targets(
            [("model.language_model.layers.29.self_attn.o_proj.weight", (4, 4))],
            ["model.layers.29.self_attn.o_proj.weight"], "t.safetensors",
            {"model.layers.29.self_attn.o_proj.weight": 16})
        self.assertEqual(got, {"model.language_model.layers.29.self_attn.o_proj.weight":
                               "model.layers.29.self_attn.o_proj.weight"})
        # …but a count mismatch stays loud, not a silent wrong-slot write.
        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit):
                voice._resolve_pack_targets(
                    [("model.language_model.layers.29.self_attn.o_proj.weight", (4, 8))],
                    ["model.layers.29.self_attn.o_proj.weight"], "t.safetensors",
                    {"model.layers.29.self_attn.o_proj.weight": 16})
        self.assertIn("not found", buf.getvalue())

class TestEntry(IsolatedVoices):
    """Launch-path proofs: the P3 split moved commands into vlib/, so entry
    identity (__main__) must be resolved explicitly, never via __file__."""

    def test_entry_script_resolves_main(self):
        import types
        fake = types.SimpleNamespace(__file__="voice.py")
        old = sys.modules.get("__main__")
        sys.modules["__main__"] = fake
        try:
            got = voice._entry_script()
        finally:
            if old is not None:
                sys.modules["__main__"] = old
        self.assertTrue(os.path.isabs(got), got)
        self.assertTrue(got.endswith("voice.py"), got)

    def test_entry_script_unknown_is_none(self):
        import types
        old = sys.modules.get("__main__")
        sys.modules["__main__"] = types.SimpleNamespace()
        try:
            self.assertIsNone(voice._entry_script())
        finally:
            if old is not None:
                sys.modules["__main__"] = old

    def test_venv_reexec_targets_entry(self):
        # Re-exec must run voice.py (absolute) with argv intact — never the module.
        import types
        d = _mktmp(self)
        vp = d / ".venv" / "bin"
        vp.mkdir(parents=True)
        (vp / "python").symlink_to(sys.executable)
        from vlib import ctx
        old_dir, old_main, old_argv = ctx.VOICES_DIR, sys.modules.get("__main__"), sys.argv[:]
        old_venv = os.environ.pop("VOICE_NO_VENV", None)
        ctx.VOICES_DIR = d
        sys.modules["__main__"] = types.SimpleNamespace(__file__="voice.py")
        sys.argv = ["voice.py", "delta", "x"]
        captured = {}
        orig_execv = os.execv

        class _Stop(BaseException):
            pass

        os.execv = lambda p, a: captured.update(path=p, argv=a) or (_ for _ in ()).throw(_Stop())
        try:
            with self.assertRaises(_Stop):
                voice._ensure_venv("delta")
        finally:
            os.execv = orig_execv
            ctx.VOICES_DIR = old_dir
            if old_main is not None:
                sys.modules["__main__"] = old_main
            sys.argv = old_argv
            if old_venv is not None:
                os.environ["VOICE_NO_VENV"] = old_venv
        self.assertTrue(captured["argv"][1].endswith("/voice.py"), captured["argv"])
        self.assertNotIn("cmds_misc", captured["argv"][1])
        self.assertEqual(captured["argv"][2:], ["delta", "x"])

    def test_entry_help_runs_main(self):
        # The suite calls cmd_* directly and never executes main()'s parser
        # wiring — a deleted flag or bad indent kills every real invocation
        # while the suite stays green. This pins the entry path end to end.
        import subprocess
        root = Path(__file__).resolve().parent.parent
        env = dict(os.environ, VOICE_NO_VENV="1")
        outs = []
        for argv in (["--help"], ["cast", "--help"], ["list"]):
            r = subprocess.run([sys.executable, str(root / "voice.py"), *argv],
                               capture_output=True, text=True, timeout=120, env=env)
            self.assertEqual(r.returncode, 0, (argv, r.stderr[-500:]))
            outs.append(r.stdout)
        self.assertIn("{get,cast,delta,graft,chorus,add,list,info,remove,cleanup,auth,report,path,doctor}", outs[0])
        self.assertIn("--speak", outs[1])

    def test_voice_tensors_coerces_bare_strings(self):
        # Hand-written voice.json may list tensors as bare strings; readers
        # must render them, never traceback (P1 crash: only.get on a str).
        got, synth = voice._voice_tensors({"tensors": ["lm_head.weight"]})
        self.assertEqual(got, [{"name": "lm_head.weight"}])
        self.assertFalse(synth)

    def test_quant_override_byte_change_fails_fast(self):
        # --quant f16 on an F32 safetensors target can never fit the fixed
        # offsets: fail up front with guidance, not warn-then-dump (B1).
        import io
        from contextlib import redirect_stdout
        d = _mktmp(self)
        write_st(d / "t.safetensors",
                 {"output.weight": (np.zeros((4, 4), np.float32), "F32")})
        write_st(d / "v.safetensors",
                 {"output.weight": (np.zeros((4, 4), np.float32), "F32")})
        header, tensors = voice.read_safetensors(str(d / "t.safetensors"))
        src = voice.open_source(str(d / "v.safetensors"))
        ref = src.ref("output.weight")
        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit):
                voice._cast_st_prepare(str(d / "t.safetensors"), tensors,
                                       src.read_f32("output.weight"), ref, {},
                                       ns(quant="f16"))
        self.assertIn("different byte size", buf.getvalue())

    def test_cancel_exits_nonzero(self):
        # Abandoned prompts (EOF/decline) must exit nonzero so scripts can
        # tell cancel from success (B2 contract).
        from unittest import mock
        from vlib.ui import _confirm
        with mock.patch("builtins.input", side_effect=EOFError), \
             mock.patch.object(sys.stdin, "isatty", return_value=True):
            with self.assertRaises(SystemExit) as cm:
                _confirm("  Overwrite? [y/N]: ",
                         ns(non_interactive=False, yes=False))
        self.assertNotEqual(cm.exception.code, 0)

    def test_requested_subset_routes_by_role(self):
        # An explicitly named tensor must land in its own slot, never the
        # output-first fallback (D2: embed request fitted onto lm_head).
        d = _mktmp(self)
        write_st(d / "v.safetensors", {
            "model.embed_tokens.weight": (np.full((4, 3), 7.0, np.float16), "F16"),
            "model.lm_head.weight": (np.full((8, 4), 8.0, np.float16), "F16")})
        write_st(d / "t.safetensors", {
            "model.embed_tokens.weight": (np.full((4, 3), 1.0, np.float16), "F16"),
            "model.lm_head.weight": (np.full((8, 4), 2.0, np.float16), "F16")})
        out = d / "o.safetensors"
        voice.cmd_cast(ns(target=str(d / "t.safetensors"), voice=str(d / "v.safetensors"),
                           out=str(out), tensors=["model.embed_tokens.weight"]))
        self.assertTrue((read_arr(out, "model.embed_tokens.weight", np.float16) == 7.0).all())
        self.assertTrue((read_arr(out, "model.lm_head.weight", np.float16) == 2.0).all())


class TestSymlinkHardening(IsolatedVoices):
    """Proof tests for the P3 symlink hardening (local CLI, self-owned dir)."""

    def test_unresolvable_path_fails_closed(self):
        # Old code swallowed resolve errors and returned (fail-open).
        with self.assertRaises(SystemExit):
            voice._assert_not_symlink("/tmp/voices-probe\x00evil")

    def test_relative_path_under_symlinked_subdir_rejected(self):
        # Old code compared raw strings, so a cwd-relative in-registry path
        # never matched the absolute VOICES_DIR scope and skipped the walk.
        import os as _os
        t = _mktmp(self)
        (t / "real" / ".voices").mkdir(parents=True)
        (t / "elsewhere").mkdir()
        try:
            (t / "real" / ".voices" / "sub").symlink_to(t / "elsewhere")
        except OSError:
            self.skipTest("symlinks unsupported here")
        ctx_mod = sys.modules["vlib.ctx"]
        old, old_ctx = voice.VOICES_DIR, ctx_mod.VOICES_DIR
        prev_cwd = _os.getcwd()
        try:
            voice.VOICES_DIR = t / "real" / ".voices"
            ctx_mod.VOICES_DIR = t / "real" / ".voices"
            _os.chdir(t)
            with self.assertRaises(SystemExit):
                voice._assert_not_symlink("real/.voices/sub/evil")
        finally:
            _os.chdir(prev_cwd)
            voice.VOICES_DIR = old
            ctx_mod.VOICES_DIR = old_ctx

    def test_scope_boundaries(self):
        # Legit nested registry path: silent. Prefix-adjacent sibling
        # (".voices2"-style, outside but string-matching): refused, as before.
        # Genuinely unrelated user path: untouched.
        (self.vdir / "myadd").mkdir()
        voice._assert_not_symlink(self.vdir / "myadd" / "voice.json")
        sib = self.vdir.parent / (self.vdir.name + "2")
        sib.mkdir(exist_ok=True)
        self.addCleanup(shutil.rmtree, sib, True)
        with self.assertRaises(SystemExit):
            voice._assert_not_symlink(sib / "voice.json")
        voice._assert_not_symlink(_mktmp(self) / "voice.json")

    def test_topup_write_refuses_symlink(self):
        d = _mktmp(self)
        hdr = {"a": {"dtype": "F16", "shape": [4], "data_offsets": [0, 8]}}
        hj = json.dumps(hdr).encode()
        real = d / "base.safetensors"
        real.write_bytes(struct.pack("<Q", len(hj)) + hj + b"\x00" * 8)
        link = d / "link.safetensors"
        try:
            link.symlink_to(real)
        except OSError:
            self.skipTest("symlinks unsupported here")
        with self.assertRaises(SystemExit):
            voice._cache_topup_write(link, [])

    def test_info_renders_chorus_provenance(self):
        import io
        import json as _json
        from contextlib import redirect_stdout
        from vlib import ctx as _ctx
        home = _mktmp(self)
        named = home / "chorus-a-b"
        named.mkdir()
        arr = np.zeros((4, 4), dtype=np.float16)
        write_st(named / "voice.safetensors",
                 {"lm_head.weight": (arr, "F16")})
        (named / "voice.json").write_text(_json.dumps({
            "name": "chorus-a-b", "type": "chorus", "method": "della",
            "pulse": "chorus-a-x", "donor": "b", "tensor_name": "lm_head.weight",
            "shape": [4, 4], "dtype": "F16",
            "config": {"density": 0.9, "epsilon": 0.05, "lambda": 1.0,
                       "rescale": "l1", "seed": 43},
            "config_hash": "abc123", "chain": ["chorus-a-x", "chorus-a-b"],
        }) + "\n")
        old, old_ctx = voice.VOICES_DIR, _ctx.VOICES_DIR
        try:
            voice.VOICES_DIR = home
            _ctx.VOICES_DIR = home
            buf = io.StringIO()
            with redirect_stdout(buf):
                voice.cmd_info(ns(voicename="chorus-a-b"))
            out = buf.getvalue()
        finally:
            voice.VOICES_DIR = old
            _ctx.VOICES_DIR = old
        for want in ("Method:  della", "Pulse:   chorus-a-x", "Donor:   b",
                     "Chain:   chorus-a-x → chorus-a-b", "density 0.9",
                     "epsilon 0.05", "seed 43", "(#abc123)"):
            self.assertIn(want, out)

    def test_info_omits_provenance_for_plain_voices(self):
        import io
        import json as _json
        from contextlib import redirect_stdout
        from vlib import ctx as _ctx
        home = _mktmp(self)
        named = home / "plain"
        named.mkdir()
        arr = np.zeros((4, 4), dtype=np.float16)
        write_st(named / "voice.safetensors",
                 {"lm_head.weight": (arr, "F16")})
        (named / "voice.json").write_text(_json.dumps({
            "name": "plain", "tensor_name": "lm_head.weight",
            "shape": [4, 4], "dtype": "F16",
        }) + "\n")
        old, old_ctx = voice.VOICES_DIR, _ctx.VOICES_DIR
        try:
            voice.VOICES_DIR = home
            _ctx.VOICES_DIR = home
            buf = io.StringIO()
            with redirect_stdout(buf):
                voice.cmd_info(ns(voicename="plain"))
            out = buf.getvalue()
        finally:
            voice.VOICES_DIR = old
            _ctx.VOICES_DIR = old
        for absent in ("Method:", "Pulse:", "Donor:", "Chain:", "Config:"):
            self.assertNotIn(absent, out)

class TestDella(unittest.TestCase):

    """Native DELLA core (vlib/della.py): prune/combine/merge on tiny fakes."""
    def test_kept_fraction_tracks_density_per_row(self):
        rng = np.random.default_rng(0)
        d = rng.standard_normal((8, 400)).astype(np.float32)
        out = voice_della.magprune(d, density=0.5, epsilon=0.1,
                                   rng=np.random.default_rng(7))
        frac = (out != 0).mean(axis=1)
        self.assertTrue(((frac > 0.40) & (frac < 0.60)).all(), frac.tolist())
        # Largest-magnitude entries survive more often than the smallest.
        mag = np.abs(d)
        kept_mag = mag[out != 0].mean()
        dropped_mag = mag[out == 0].mean()
        self.assertGreater(kept_mag, dropped_mag)

    def test_seeded_run_bit_identical(self):
        rng = np.random.default_rng(1)
        d = rng.standard_normal((6, 64)).astype(np.float32)
        a = voice_della.magprune(d, density=0.8, epsilon=0.1,
                                 rng=np.random.default_rng(42))
        b = voice_della.magprune(d, density=0.8, epsilon=0.1,
                                 rng=np.random.default_rng(42))
        self.assertTrue(np.array_equal(a, b))
        base = rng.standard_normal((9, 32)).astype(np.float32)
        vs = [base + 1, base - 1]
        m1 = voice_della.merge(base, vs, density=0.8, epsilon=0.1, seed=42)
        m2 = voice_della.merge(base, vs, density=0.8, epsilon=0.1, seed=42)
        self.assertTrue(np.array_equal(m1, m2))
        m3 = voice_della.merge(base, vs, density=0.8, epsilon=0.1, seed=43)
        self.assertFalse(np.array_equal(m1, m3))
        mb1 = voice_della.merge(base, vs, density=0.8, epsilon=0.1, seed=42,
                                block_rows=4096)
        mb2 = voice_della.merge(base, vs, density=0.8, epsilon=0.1, seed=42,
                                block_rows=2)
        self.assertTrue(np.array_equal(mb1, mb2))

    def test_density_epsilon_violations_raise(self):
        d = np.ones((2, 8), dtype=np.float32)
        for density, epsilon in ((0.9, 0.2), (0.05, 0.1)):
            with self.assertRaises(ValueError):
                voice_della.magprune(d, density=density, epsilon=epsilon)
        with self.assertRaises(ValueError):
            voice_della.merge(d, [d + 1], density=0.95, epsilon=0.1, seed=0)

    def test_noop_and_zero_density(self):
        d = np.ones((2, 8), dtype=np.float32)
        self.assertTrue(np.array_equal(
            voice_della.magprune(d, density=1.0, epsilon=0.1), d))
        self.assertTrue((voice_della.magprune(d, density=0.0, epsilon=0.1) == 0).all())

    def test_empty_voice_list_returns_base(self):
        base = np.ones((3, 5), dtype=np.float32)
        out = voice_della.merge(base, [])
        self.assertTrue(np.array_equal(out, base))

    def test_f16_in_base_dtype_out(self):
        rng = np.random.default_rng(2)
        base = (rng.standard_normal((4, 16)) * 2).astype(np.float16)
        donor = (base.astype(np.float32) + 1).astype(np.float16)
        out = voice_della.merge(base, [donor], density=0.8, epsilon=0.1, seed=1)
        self.assertEqual(out.dtype, np.float16)
        self.assertTrue(np.isfinite(out.astype(np.float32)).all())

    def test_all_pruned_row_stays_finite(self):
        tiny = np.full((2, 50), 1e-9, dtype=np.float32)
        out = voice_della.magprune(tiny, density=0.05, epsilon=0.04,
                                   rng=np.random.default_rng(0))
        self.assertTrue(np.isfinite(out).all())

    def test_elected_combine_matches_upstream_formula(self):
        rng = np.random.default_rng(3)
        base = rng.standard_normal((5, 40)).astype(np.float32)
        v1, v2 = base + 1, base - 1
        pruned = [voice_della.magprune(v - base, density=0.8, epsilon=0.1,
                                       rng=np.random.default_rng(5)) for v in (v1, v2)]
        got = voice_della.combine(base, pruned, [1.0, 1.0])
        stacked = np.stack([p.astype(np.float32) for p in pruned])
        majority = np.where(stacked.sum(0) >= 0, 1.0, -1.0)
        elected = np.sign(stacked) == majority
        mixed = np.where(elected, stacked, 0).sum(0)
        divisor = np.where(elected, 1.0, 0).sum(0)
        divisor[divisor == 0] = 1
        self.assertTrue(np.allclose(got, base + mixed / divisor, atol=1e-6))

    def test_per_row_l1_rescale_not_global(self):
        # Per-row L1 (vlib fork) keeps each row's own sum|x|; a whole-tensor
        # L1 (upstream) would scale every row by one global factor instead.
        rng = np.random.default_rng(4)
        scales = np.array([0.1, 1, 5, 0.5, 2, 10], dtype=np.float32)[:, None]
        d = (rng.standard_normal((6, 100)).astype(np.float32)) * scales
        out = voice_della.magprune(d, density=0.7, epsilon=0.1,
                                   rng=np.random.default_rng(11))
        before = np.abs(d).sum(axis=-1)
        after = np.abs(out).sum(axis=-1)
        live = after >= 1e-7
        self.assertTrue(live.all())
        self.assertTrue(np.allclose(after[live], before[live], rtol=1e-5))

if __name__ == "__main__":
    unittest.main()