"""v0.71.36 — Data Moat II: semantic layer + canaries + replay."""

from __future__ import annotations

import json
import re

import pytest

_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")


def _clean(text: str) -> str:
    """Strip ANSI + collapse whitespace.

    Rich splits flag names with color codes and wraps at terminal width;
    asserting on raw output has broken CI three times (v0.71.26/32/35).
    """
    return " ".join(_ANSI_RE.sub("", text).split())


class TestResolvePooling:
    def test_allowlisted_model_short_circuits(self, monkeypatch):
        from soup_cli.utils import embed

        def _no_network(mid):
            raise AssertionError("should not reach network")

        monkeypatch.setattr(embed, "_fetch_pooling_config", _no_network)

        assert embed.resolve_pooling("sentence-transformers/all-MiniLM-L6-v2") == "mean"
        assert embed.resolve_pooling("sentence-transformers/all-mpnet-base-v2") == "mean"

    def test_allowlist_is_case_insensitive(self, monkeypatch):
        from soup_cli.utils import embed

        def _no_network(mid):
            raise AssertionError("should not reach network")

        monkeypatch.setattr(embed, "_fetch_pooling_config", _no_network)

        assert embed.resolve_pooling("Sentence-Transformers/All-MiniLM-L6-v2") == "mean"

    def test_cls_pooling_config_is_refused(self, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed,
            "_fetch_pooling_config",
            lambda mid: {
                "pooling_mode_mean_tokens": False,
                "pooling_mode_cls_token": True,
            },
        )
        with pytest.raises(ValueError, match="cls"):
            embed.resolve_pooling("org/some-encoder")

    def test_cls_precedence_over_contradictory_mean_flag(self, monkeypatch):
        """_NON_MEAN_POOLING_KEYS is checked BEFORE pooling_mode_mean_tokens.

        A self-contradictory config that sets both the cls flag AND the mean
        flag true must still be refused — and refused for being cls, not
        silently accepted because the mean flag also happens to be true.
        Pins the check ordering so a future refactor that collapses this
        into one flat check can't ship silently.
        """
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed,
            "_fetch_pooling_config",
            lambda mid: {
                "pooling_mode_mean_tokens": True,
                "pooling_mode_cls_token": True,
            },
        )
        with pytest.raises(ValueError, match="cls"):
            embed.resolve_pooling("org/contradictory-encoder")

    def test_mean_pooling_config_is_accepted(self, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed,
            "_fetch_pooling_config",
            lambda mid: {
                "pooling_mode_mean_tokens": True,
                "pooling_mode_cls_token": False,
            },
        )
        assert embed.resolve_pooling("some/mean-model") == "mean"

    def test_unfetchable_config_is_refused_not_assumed(self, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(embed, "_fetch_pooling_config", lambda mid: None)
        with pytest.raises(ValueError, match="cannot verify pooling"):
            embed.resolve_pooling("some/unknown-model")

    def test_refusal_names_what_it_found(self, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed,
            "_fetch_pooling_config",
            lambda mid: {
                "pooling_mode_mean_tokens": False,
                "pooling_mode_max_tokens": True,
            },
        )
        with pytest.raises(ValueError) as exc:
            embed.resolve_pooling("org/other-encoder")
        assert "max" in str(exc.value)

    @pytest.mark.parametrize("bad", [None, 123, True, 1.5, b"bytes"])
    def test_non_string_model_id_rejected(self, bad):
        from soup_cli.utils.embed import resolve_pooling

        with pytest.raises(TypeError, match="must be str"):
            resolve_pooling(bad)

    # Each guard below asserts its OWN message. A bare
    # `pytest.raises((ValueError, TypeError))` was vacuous here: with the
    # empty-check deleted, "" falls through to the FALLBACK refusal
    # ("cannot verify pooling") and still raises ValueError, so the broad
    # form could not tell the two apart — proven by mutation.
    @pytest.mark.parametrize("bad", ["", "   ", "\t\n"])
    def test_empty_model_id_names_the_empty_guard(self, bad, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed, "_fetch_pooling_config",
            lambda mid: pytest.fail("must reject before any fetch"),
        )
        with pytest.raises(ValueError, match="non-empty"):
            embed.resolve_pooling(bad)

    def test_null_byte_model_id_names_the_null_guard(self, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed, "_fetch_pooling_config",
            lambda mid: pytest.fail("must reject before any fetch"),
        )
        with pytest.raises(ValueError, match="null byte"):
            embed.resolve_pooling("org/model\x00evil")

    def test_overlong_model_id_names_the_length_guard(self, monkeypatch):
        from soup_cli.utils import embed

        monkeypatch.setattr(
            embed, "_fetch_pooling_config",
            lambda mid: pytest.fail("must reject before any fetch"),
        )
        with pytest.raises(ValueError, match="too long"):
            embed.resolve_pooling("o/" + "x" * (embed._MAX_MODEL_ID_CHARS + 1))

    def test_model_id_is_stripped_not_just_validated(self):
        from soup_cli.utils.embed import resolve_pooling

        # surrounding whitespace must not defeat the allowlist
        assert resolve_pooling(
            "  sentence-transformers/all-MiniLM-L6-v2  "
        ) == "mean"


class TestEmbedTexts:
    def test_rejects_over_cap(self):
        from soup_cli.utils.embed import _MAX_ROWS, embed_texts

        with pytest.raises(ValueError, match="too many texts"):
            embed_texts(["x"] * (_MAX_ROWS + 1))

    def test_rejects_empty_list(self):
        from soup_cli.utils.embed import embed_texts

        with pytest.raises(ValueError, match="at least one text"):
            embed_texts([])

    def test_rejects_non_string_row(self):
        from soup_cli.utils.embed import embed_texts

        with pytest.raises(TypeError, match=r"texts\[1\] must be str"):
            embed_texts(["ok", 42])

    @pytest.mark.parametrize("bad", [0, -1, True])
    def test_rejects_bad_batch_size(self, bad):
        from soup_cli.utils.embed import embed_texts

        with pytest.raises((ValueError, TypeError)):
            embed_texts(["x"], batch_size=bad)

    def test_validate_texts_truncates_long_rows(self):
        from soup_cli.utils.embed import _MAX_CHARS_PER_ROW, _validate_texts

        out = _validate_texts(["a" * (_MAX_CHARS_PER_ROW + 500)])
        assert len(out[0]) == _MAX_CHARS_PER_ROW

    def test_validate_texts_rejects_bare_string(self):
        """A bare str is a sequence of chars — almost always a caller bug."""
        from soup_cli.utils.embed import _validate_texts

        with pytest.raises(TypeError, match="sequence of str"):
            _validate_texts("not a list")

    def test_mean_pool_masks_padding(self):
        """Padded positions must NOT contribute to the mean."""
        import numpy as np

        from soup_cli.utils.embed import _mean_pool

        torch = pytest.importorskip("torch")
        # 3 tokens; the last is padding carrying a huge value that would
        # wreck an unmasked mean (999 -> ~334 instead of 2.0).
        hidden = torch.tensor([[[1.0, 1.0], [3.0, 3.0], [999.0, 999.0]]])
        mask = torch.tensor([[1, 1, 0]])
        out = _mean_pool(hidden, mask).numpy()
        np.testing.assert_allclose(out, np.array([[2.0, 2.0]]), rtol=1e-6)

    def test_mean_pool_all_padding_does_not_divide_by_zero(self):
        import numpy as np

        from soup_cli.utils.embed import _mean_pool

        torch = pytest.importorskip("torch")
        hidden = torch.tensor([[[5.0, 5.0]]])
        mask = torch.tensor([[0]])
        out = _mean_pool(hidden, mask).numpy()
        assert np.isfinite(out).all()

    def test_l2_normalize_rows(self):
        import numpy as np

        from soup_cli.utils.embed import _l2_normalize

        vecs = np.array([[3.0, 4.0], [0.0, 0.0]], dtype=np.float32)
        out = _l2_normalize(vecs)
        np.testing.assert_allclose(np.linalg.norm(out[0]), 1.0, rtol=1e-6)
        # a zero vector must not become NaN
        assert np.isfinite(out[1]).all()

    def test_l2_normalize_makes_cosine_a_dot_product(self):
        import numpy as np

        from soup_cli.utils.embed import _l2_normalize

        vecs = _l2_normalize(np.array([[2.0, 0.0], [0.0, 5.0]], dtype=np.float32))
        assert float(vecs[0] @ vecs[1]) == pytest.approx(0.0, abs=1e-6)
        assert float(vecs[0] @ vecs[0]) == pytest.approx(1.0, abs=1e-6)

    def test_refuses_unverified_model_before_any_download(self, monkeypatch):
        """resolve_pooling must gate embed_texts BEFORE a model is fetched."""
        from soup_cli.utils import embed

        monkeypatch.setattr(embed, "_fetch_pooling_config", lambda mid: None)
        with pytest.raises(ValueError, match="cannot verify pooling"):
            embed.embed_texts(["hello"], model_id="org/unverified-encoder")


class TestGreedySemdedup:
    def _vecs(self, rows):
        import numpy as np

        arr = np.array(rows, dtype=np.float32)
        norms = np.linalg.norm(arr, axis=1, keepdims=True)
        return arr / np.where(norms < 1e-12, 1.0, norms)

    def test_identical_vectors_collapse_to_one(self):
        from soup_cli.utils.semdedup import greedy_semdedup

        vecs = self._vecs([[1.0, 0.0], [1.0, 0.0], [1.0, 0.0]])
        rep = greedy_semdedup(vecs, threshold=0.9)
        assert rep.kept == (0,)
        assert rep.dropped == (1, 2)

    def test_orthogonal_vectors_all_kept(self):
        from soup_cli.utils.semdedup import greedy_semdedup

        vecs = self._vecs([[1.0, 0.0], [0.0, 1.0]])
        rep = greedy_semdedup(vecs, threshold=0.9)
        assert rep.kept == (0, 1)
        assert rep.dropped == ()

    def _exact_half_cosine(self):
        """Two UNIT vectors whose cosine is EXACTLY 0.5 in float32.

        The obvious construction — arccos(0.8) then cos/sin — yields
        0.800000011920929, i.e. strictly ABOVE the threshold, so `>=` and
        `>` behave identically and the boundary is never actually tested.
        Here x=0.5 is dyadic (exact in binary) and the first vector's y is
        exactly 0.0, so the dot product is exactly 0.5 with no rounding.
        """
        import numpy as np

        return np.array([[1.0, 0.0], [0.5, np.sqrt(3) / 2]], dtype=np.float32)

    def test_boundary_fixture_is_exact(self):
        """Guard the guard: if this drifts, the boundary tests go vacuous."""
        import numpy as np

        vecs = self._exact_half_cosine()
        assert float(vecs[0] @ vecs[1]) == 0.5
        assert float(np.linalg.norm(vecs[1])) == pytest.approx(1.0, abs=1e-6)

    def test_threshold_boundary_is_inclusive(self):
        """cosine == threshold must DROP (>=), not keep."""
        from soup_cli.utils.semdedup import greedy_semdedup

        rep = greedy_semdedup(self._exact_half_cosine(), threshold=0.5)
        assert rep.dropped == (1,), "cosine == threshold must drop"

    def test_just_below_threshold_is_kept(self):
        from soup_cli.utils.semdedup import greedy_semdedup

        rep = greedy_semdedup(self._exact_half_cosine(), threshold=0.51)
        assert rep.dropped == ()

    def test_pairs_record_which_kept_row_and_cosine(self):
        from soup_cli.utils.semdedup import greedy_semdedup

        vecs = self._vecs([[1.0, 0.0], [0.0, 1.0], [1.0, 0.0]])
        rep = greedy_semdedup(vecs, threshold=0.9)
        assert len(rep.pairs) == 1
        dropped_idx, kept_idx, cosine = rep.pairs[0]
        assert (dropped_idx, kept_idx) == (2, 0)
        assert cosine == pytest.approx(1.0, abs=1e-5)

    def test_pairs_name_the_nearest_kept_row_not_just_any(self):
        """The provenance must identify WHICH kept row it collided with.

        Row 2 is near row 1 and far from row 0; a report that blindly cited
        kept[0] would pass a weaker test.
        """
        import numpy as np

        from soup_cli.utils.semdedup import greedy_semdedup

        theta = float(np.arccos(0.95))
        vecs = self._vecs([
            [1.0, 0.0],
            [0.0, 1.0],
            [float(np.sin(theta)), float(np.cos(theta))],
        ])
        rep = greedy_semdedup(vecs, threshold=0.9)
        assert rep.pairs[0][0] == 2
        assert rep.pairs[0][1] == 1, "must cite the NEAREST kept row"

    def test_transitive_chain_keeps_first_only(self):
        """A~B, B~C but A!~C: greedy keeps A, drops B; C compared to A only."""
        import numpy as np

        from soup_cli.utils.semdedup import greedy_semdedup

        t1 = float(np.arccos(0.95))
        vecs = self._vecs([
            [1.0, 0.0],
            [float(np.cos(t1)), float(np.sin(t1))],
            [float(np.cos(2 * t1)), float(np.sin(2 * t1))],
        ])
        rep = greedy_semdedup(vecs, threshold=0.9)
        assert rep.kept == (0, 2)
        assert rep.dropped == (1,)

    def test_single_row(self):
        from soup_cli.utils.semdedup import greedy_semdedup

        rep = greedy_semdedup(self._vecs([[1.0, 0.0]]), threshold=0.9)
        assert rep.kept == (0,)

    def test_empty_input(self):
        import numpy as np

        from soup_cli.utils.semdedup import greedy_semdedup

        rep = greedy_semdedup(
            np.zeros((0, 2), dtype=np.float32), threshold=0.9
        )
        assert rep.kept == ()
        assert rep.dropped == ()

    def test_rejects_non_2d(self):
        import numpy as np

        from soup_cli.utils.semdedup import greedy_semdedup

        with pytest.raises(ValueError, match="2-D"):
            greedy_semdedup(np.zeros((5,), dtype=np.float32), threshold=0.9)

    def test_rejects_over_cap(self):
        import numpy as np

        from soup_cli.utils.semdedup import _MAX_SEMDEDUP_ROWS, greedy_semdedup

        vecs = np.zeros((_MAX_SEMDEDUP_ROWS + 1, 2), dtype=np.float32)
        with pytest.raises(ValueError, match="too many rows"):
            greedy_semdedup(vecs, threshold=0.9)

    @pytest.mark.parametrize("bad", [-0.1, 1.1, float("nan"), True, "x"])
    def test_rejects_bad_threshold(self, bad):
        from soup_cli.utils.semdedup import greedy_semdedup

        with pytest.raises((ValueError, TypeError)):
            greedy_semdedup(self._vecs([[1.0, 0.0]]), threshold=bad)

    def test_kept_and_dropped_partition_the_input(self):
        from soup_cli.utils.semdedup import greedy_semdedup

        vecs = self._vecs(
            [[1.0, 0.0], [1.0, 0.0], [0.0, 1.0], [0.0, 1.0], [1.0, 1.0]]
        )
        rep = greedy_semdedup(vecs, threshold=0.9)
        assert sorted(rep.kept + rep.dropped) == list(range(5))
        assert not set(rep.kept) & set(rep.dropped)

    def test_report_is_frozen(self):
        import dataclasses

        from soup_cli.utils.semdedup import DedupReport

        rep = DedupReport(kept=(0,), dropped=(), pairs=(), threshold=0.9)
        with pytest.raises(dataclasses.FrozenInstanceError):
            rep.threshold = 0.5


class TestExtrasHintsAreEscaped:
    """Rich eats `[extra]` as a markup tag unless it is escaped.

    Unescaped, `pip install 'soup-cli[eval]'` renders as
    `pip install 'soup-cli'` -- which installs the base package WITHOUT the
    extra the user is missing, so following the hint appears to succeed and
    the feature still fails. Found during v0.71.36; same class as the
    v0.71.28 \\[mcp] fix, which only fixed its own site.

    Sites that do NOT go through Rich (plain exception text, docstrings)
    must NOT be escaped -- a backslash would show up literally.
    """

    # Rich style tags. A hint string carrying one of these is console.print
    # output; a bare `raise ImportError("... soup-cli[mlx]")` is NOT and must
    # be left alone (escaping it would print a literal backslash).
    _RICH_TAGS = ("[/]", "[bold]", "[red]", "[yellow]", "[green]", "[dim]")

    def test_rich_renders_unescaped_bracket_away(self):
        """Pin the underlying Rich behaviour this whole class exists for."""
        from io import StringIO

        from rich.console import Console

        def render(markup):
            buf = StringIO()
            Console(file=buf, force_terminal=False, width=100).print(markup)
            return buf.getvalue().strip()

        assert render("[bold]pip install 'soup-cli[train]'[/]") == (
            "pip install 'soup-cli'"
        ), "unescaped bracket must be eaten (this is the bug)"
        assert render("[bold]pip install 'soup-cli\\[train]'[/]") == (
            "pip install 'soup-cli[train]'"
        ), "escaped bracket must survive (this is the fix)"

    def test_no_unescaped_extras_hint_in_rich_markup(self):
        """Every hint carrying Rich markup must escape its bracket."""
        import pathlib
        import re

        import soup_cli

        root = pathlib.Path(soup_cli.__file__).parent
        bad_re = re.compile(r"(?<!\\)soup-cli\[[a-z][a-z0-9-]*\]")
        offenders = []
        for path in sorted(root.rglob("*.py")):
            rel = path.relative_to(root).as_posix()
            for lineno, line in enumerate(
                path.read_text(encoding="utf-8").splitlines(), start=1
            ):
                if line.strip().startswith("#"):
                    continue  # a code comment never reaches Rich
                if not any(tag in line for tag in self._RICH_TAGS):
                    continue  # not Rich output — must NOT be escaped
                if bad_re.search(line):
                    offenders.append(f"{rel}:{lineno}: {line.strip()}")
        assert not offenders, (
            "unescaped soup-cli[extra] inside Rich markup — the bracket is "
            "eaten and the printed command installs WITHOUT the extra:\n"
            + "\n".join(offenders)
        )

    @pytest.mark.parametrize(
        "argv,extra",
        [
            (["data", "dedup", "--help"], "soup-cli[train]"),
            (["train", "--help"], "soup-cli[carbon]"),
        ],
    )
    def test_typer_help_keeps_the_extra_bracket(self, argv, extra):
        """Typer help is Rich-rendered too — the bracket must survive.

        `--semantic` help shipped as "Requires soup-cli." before this fix.
        """
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner(env={"COLUMNS": "200"}).invoke(app, argv)
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert extra in _clean(res.output), (
            f"{extra!r} missing from `{' '.join(argv)}` — Rich ate the bracket"
        )

    def test_plain_exception_hints_are_not_escaped(self):
        """The counter-rule: non-Rich text must NOT gain a backslash.

        `raise ImportError('... pip install "soup-cli[mlx]"')` never reaches
        Rich, so escaping it would surface a literal backslash to the user.

        The quoting itself is v0.71.37's business (cmd.exe cannot strip `'`);
        this test only cares that the bracket stays bare here.
        """
        import pathlib

        import soup_cli

        root = pathlib.Path(soup_cli.__file__).parent
        text = (root / "trainer" / "mlx_sft.py").read_text(encoding="utf-8")
        # Quote-agnostic on purpose: the shell quoting around the name is
        # v0.71.37's concern, this rule is only about the bracket staying bare.
        assert "soup-cli[mlx]" in text
        assert "soup-cli\\[mlx]" not in text


class TestDedupSemanticCli:
    def _write(self, tmp_path, rows):
        path = tmp_path / "data.jsonl"
        path.write_text(
            "\n".join(json.dumps(r) for r in rows), encoding="utf-8"
        )
        return path

    def test_help_mentions_semantic(self):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner().invoke(app, ["data", "dedup", "--help"])
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert "semantic" in _clean(res.output)

    def test_semantic_dedups_via_injected_embedder(self, tmp_path, monkeypatch):
        """CLI wiring tested without a model download."""
        import numpy as np
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data as data_cmd

        rows = [
            {"text": "the cat sat on the mat"},
            {"text": "a feline rested upon the rug"},
            {"text": "quantum chromodynamics is hard"},
        ]
        path = self._write(tmp_path, rows)
        # rows 0 and 1 are "paraphrases" -> near-identical vectors
        fake = np.array(
            [[1.0, 0.0], [0.999, 0.0447], [0.0, 1.0]], dtype=np.float32
        )
        monkeypatch.setattr(data_cmd, "embed_texts", lambda *a, **k: fake)
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app,
            ["data", "dedup", str(path), "--semantic",
             "--threshold", "0.9", "-o", "out.jsonl"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        kept = [
            json.loads(ln)
            for ln in (tmp_path / "out.jsonl").read_text().splitlines() if ln
        ]
        assert len(kept) == 2
        assert kept[0]["text"] == "the cat sat on the mat"
        assert kept[1]["text"] == "quantum chromodynamics is hard"

    def test_semantic_does_not_require_datasketch(self, tmp_path, monkeypatch):
        """--semantic must work for a user with [train] but NOT [data].

        The datasketch import used to sit at the TOP of dedup(), before the
        file check, so the semantic path would have died with 'datasketch
        not installed' despite never needing MinHash.
        """
        import builtins

        import numpy as np
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data as data_cmd

        real_import = builtins.__import__

        def _no_datasketch(name, *args, **kwargs):
            if name == "datasketch":
                raise ImportError("No module named 'datasketch'")
            return real_import(name, *args, **kwargs)

        monkeypatch.setattr(builtins, "__import__", _no_datasketch)
        monkeypatch.setattr(
            data_cmd, "embed_texts",
            lambda *a, **k: np.array(
                [[1.0, 0.0], [0.0, 1.0]], dtype=np.float32
            ),
        )
        path = self._write(tmp_path, [{"text": "a"}, {"text": "b"}])
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "dedup", str(path), "--semantic", "-o", "out.jsonl"]
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert "datasketch" not in _clean(res.output)

    def test_semantic_import_error_is_friendly(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data as data_cmd

        path = self._write(tmp_path, [{"text": "a"}])

        def _boom(*a, **k):
            raise ImportError("No module named 'torch'")

        monkeypatch.setattr(data_cmd, "embed_texts", _boom)
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "dedup", str(path), "--semantic"]
        )
        assert res.exit_code == 1
        assert "soup-cli[train]" in _clean(res.output)

    def test_semantic_pooling_refusal_surfaces(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data as data_cmd

        path = self._write(tmp_path, [{"text": "a"}])

        def _refuse(*a, **k):
            raise ValueError("cannot verify pooling for 'x/y'")

        monkeypatch.setattr(data_cmd, "embed_texts", _refuse)
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "dedup", str(path), "--semantic"]
        )
        assert res.exit_code == 1
        assert "pooling" in _clean(res.output)

    def test_minhash_path_untouched_without_flag(self, tmp_path, monkeypatch):
        """--semantic absent -> embed_texts must never be called."""
        pytest.importorskip("datasketch")
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data as data_cmd

        called = []
        monkeypatch.setattr(
            data_cmd, "embed_texts",
            lambda *a, **k: called.append(1) or None,
        )
        path = self._write(
            tmp_path, [{"text": "a b c"}, {"text": "x y z"}]
        )
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(app, ["data", "dedup", str(path)])
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert called == [], "MinHash path must not touch the embedder"

    def test_field_selects_what_is_embedded(self, tmp_path, monkeypatch):
        """--field composes with --semantic (same meaning as for MinHash)."""
        import numpy as np
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data as data_cmd

        seen = {}

        def _capture(texts, **kwargs):
            seen["texts"] = list(texts)
            return np.array([[1.0, 0.0], [0.0, 1.0]], dtype=np.float32)

        monkeypatch.setattr(data_cmd, "embed_texts", _capture)
        rows = [
            {"text": "keep me", "noise": "IGNORED"},
            {"text": "other", "noise": "IGNORED"},
        ]
        path = self._write(tmp_path, rows)
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app,
            ["data", "dedup", str(path), "--semantic",
             "--field", "text", "-o", "out.jsonl"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert seen["texts"] == ["keep me", "other"]
        assert not any("IGNORED" in t for t in seen["texts"])

    def test_output_outside_cwd_rejected(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        path = self._write(tmp_path, [{"text": "a"}])
        work = tmp_path / "work"
        work.mkdir()
        monkeypatch.chdir(work)
        res = CliRunner().invoke(
            app,
            ["data", "dedup", str(path), "--semantic",
             "-o", str(tmp_path / "escape.jsonl")],
        )
        assert res.exit_code == 1
        assert "outside" in _clean(res.output).lower()


class TestResolveK:
    @pytest.mark.parametrize(
        "n_rows,expected",
        [(1, 1), (3, 1), (4, 2), (8, 2), (200, 10), (1000, 22), (100_000, 25)],
    )
    def test_auto_heuristic(self, n_rows, expected):
        from soup_cli.utils.topics import resolve_k

        assert resolve_k(n_rows, "auto") == expected

    def test_auto_is_capped_so_the_table_stays_one_screen(self):
        from soup_cli.utils.topics import _MAX_K, resolve_k

        assert resolve_k(10_000_000, "auto") == _MAX_K

    def test_explicit_k_used_as_is(self):
        from soup_cli.utils.topics import resolve_k

        assert resolve_k(1000, 7) == 7

    def test_explicit_k_is_not_capped_by_max_k(self):
        """--clusters is the operator's call; only 'auto' is capped."""
        from soup_cli.utils.topics import _MAX_K, resolve_k

        assert resolve_k(1000, _MAX_K + 5) == _MAX_K + 5

    def test_explicit_k_clamped_to_n_rows(self):
        from soup_cli.utils.topics import resolve_k

        assert resolve_k(3, 10) == 3

    def test_auto_is_case_and_space_insensitive(self):
        from soup_cli.utils.topics import resolve_k

        assert resolve_k(200, " AUTO ") == 10

    @pytest.mark.parametrize("bad", [0, -1, True, "seven", 1.5])
    def test_rejects_bad_k(self, bad):
        from soup_cli.utils.topics import resolve_k

        with pytest.raises((ValueError, TypeError)):
            resolve_k(100, bad)


class TestKmeans:
    def _blobs(self):
        import numpy as np

        rng = np.random.default_rng(0)
        left = rng.normal([-5.0, 0.0], 0.1, size=(20, 2))
        right = rng.normal([5.0, 0.0], 0.1, size=(20, 2))
        return np.vstack([left, right]).astype(np.float32)

    def test_separates_two_obvious_blobs(self):
        from soup_cli.utils.topics import kmeans

        labels = kmeans(self._blobs(), k=2, seed=0)
        assert len(set(labels[:20].tolist())) == 1
        assert len(set(labels[20:].tolist())) == 1
        assert labels[0] != labels[20]

    def _unstructured(self):
        """Points with NO cluster structure, so k-means++ init decides.

        Two well-separated blobs converge to the same partition from any
        init, which makes the seed untestable on them (verified: 6 distinct
        seeds -> 2 partitions, and an unseeded RNG passes a same-seed
        equality check by luck). Unstructured points give 6 distinct
        partitions for 6 seeds, so ignoring the seed is detectable.
        """
        import numpy as np

        return np.random.default_rng(0).random((60, 4)).astype(np.float32)

    def test_deterministic_by_seed(self):
        import numpy as np

        from soup_cli.utils.topics import kmeans

        vecs = self._unstructured()
        first = kmeans(vecs, k=5, seed=42)
        for _ in range(3):
            np.testing.assert_array_equal(
                first, kmeans(vecs, k=5, seed=42),
                err_msg="same seed must give the same partition",
            )

    def test_seed_actually_influences_the_result(self):
        """Guard the guard: if seeds stopped mattering, the test above
        would pass even with the seed ignored."""
        from soup_cli.utils.topics import kmeans

        vecs = self._unstructured()
        partitions = {
            tuple(kmeans(vecs, k=5, seed=seed).tolist()) for seed in range(6)
        }
        assert len(partitions) > 1, (
            "seeds produce identical partitions on this fixture — it cannot "
            "detect an unseeded RNG"
        )

    def test_k_equals_one(self):
        from soup_cli.utils.topics import kmeans

        labels = kmeans(self._blobs(), k=1, seed=0)
        assert set(labels.tolist()) == {0}

    def test_k_greater_than_n_is_clamped(self):
        import numpy as np

        from soup_cli.utils.topics import kmeans

        vecs = np.array([[0.0, 0.0], [1.0, 1.0]], dtype=np.float32)
        labels = kmeans(vecs, k=5, seed=0)
        assert len(labels) == 2
        assert set(labels.tolist()) <= {0, 1}

    def test_labels_are_in_range(self):
        from soup_cli.utils.topics import kmeans

        labels = kmeans(self._blobs(), k=3, seed=1)
        assert all(0 <= int(x) < 3 for x in labels)

    def test_empty_input(self):
        import numpy as np

        from soup_cli.utils.topics import kmeans

        labels = kmeans(np.zeros((0, 2), dtype=np.float32), k=2, seed=0)
        assert len(labels) == 0

    def test_rejects_non_2d(self):
        import numpy as np

        from soup_cli.utils.topics import kmeans

        with pytest.raises(ValueError, match="2-D"):
            kmeans(np.zeros((5,), dtype=np.float32), k=2, seed=0)

    def test_rejects_over_cap(self):
        import numpy as np

        from soup_cli.utils.topics import _MAX_TOPIC_ROWS, kmeans

        vecs = np.zeros((_MAX_TOPIC_ROWS + 1, 2), dtype=np.float32)
        with pytest.raises(ValueError, match="too many rows"):
            kmeans(vecs, k=2, seed=0)

    def test_identical_points_do_not_crash(self):
        """All-duplicate input makes every k-means++ distance zero."""
        import numpy as np

        from soup_cli.utils.topics import kmeans

        vecs = np.ones((10, 3), dtype=np.float32)
        labels = kmeans(vecs, k=3, seed=0)
        assert len(labels) == 10


class TestCtfidfLabels:
    def _global_vs_specific(self):
        """5 clusters, each ``["the", "the", "term_i"]``.

        The filler word is deliberately MORE frequent in-cluster than the
        specific term, so plain per-cluster TF picks "the" every time and
        ONLY the inverse-cluster-frequency term can demote it. The obvious
        fixture (specific term more frequent than the filler) is vacuous:
        plain TF already picks the specific term, so deleting the idf
        changes nothing.

        The margin is arithmetic, not luck: the idf ratio at 5 clusters is
        log(1+5/1)/log(1+5/5) = 2.58, and tf("the")/tf(term) = 2.0 < 2.58.
        """
        docs = [["the", "the", f"term{i}"] for i in range(5)]
        return docs, list(range(5))

    def test_cluster_specific_term_beats_global_term(self):
        """The property that makes c-TF-IDF != plain TF-IDF."""
        from soup_cli.utils.topics import ctfidf_labels

        docs, labels = self._global_vs_specific()
        out = ctfidf_labels(docs, labels, k=5, top_n=1)
        assert out == [(f"term{i}",) for i in range(5)], (
            "a term unique to one cluster must outrank a filler word that is "
            "more frequent but present in every cluster"
        )

    def test_the_filler_is_the_plain_tf_winner(self):
        """Guard the guard: prove plain TF really would pick the filler.

        If this ever fails, the fixture stopped discriminating and the test
        above would pass even with the idf term deleted.
        """
        from collections import Counter

        docs, _ = self._global_vs_specific()
        counts = Counter(docs[0])
        assert counts["the"] > counts["term0"]

    def test_top_n_respected(self):
        from soup_cli.utils.topics import ctfidf_labels

        out = ctfidf_labels(
            [["alpha", "beta", "gamma"], ["delta"]], [0, 1], k=2, top_n=2
        )
        assert len(out[0]) <= 2

    def test_empty_cluster_yields_empty_terms(self):
        from soup_cli.utils.topics import ctfidf_labels

        out = ctfidf_labels([["a"]], [0], k=3, top_n=2)
        assert out[1] == ()
        assert out[2] == ()

    def test_length_mismatch_rejected(self):
        from soup_cli.utils.topics import ctfidf_labels

        with pytest.raises(ValueError, match="same length"):
            ctfidf_labels([["a"], ["b"]], [0], k=1, top_n=1)

    def test_out_of_range_label_ignored(self):
        from soup_cli.utils.topics import ctfidf_labels

        out = ctfidf_labels([["a"], ["b"]], [0, 99], k=1, top_n=1)
        assert out[0] == ("a",)

    @pytest.mark.parametrize("bad", [0, -1, True])
    def test_rejects_bad_top_n(self, bad):
        from soup_cli.utils.topics import ctfidf_labels

        with pytest.raises((ValueError, TypeError)):
            ctfidf_labels([["a"]], [0], k=1, top_n=bad)


class TestBuildTopicReport:
    def _rows(self, tag, count):
        return [
            {"messages": [{"role": "assistant", "content": f"{tag} {i}"}]}
            for i in range(count)
        ]

    def test_fractions_sum_to_one(self):
        from soup_cli.utils.topics import build_topic_report

        rows = self._rows("python code", 6) + self._rows("protein fold", 4)
        rep = build_topic_report(rows, [0] * 6 + [1] * 4, k=2)
        assert rep.n_rows == 10
        assert sum(t.fraction for t in rep.topics) == pytest.approx(1.0)
        assert {t.size for t in rep.topics} == {6, 4}

    def test_topics_sorted_by_coverage_desc(self):
        from soup_cli.utils.topics import build_topic_report

        rows = self._rows("small", 2) + self._rows("big", 8)
        rep = build_topic_report(rows, [0] * 2 + [1] * 8, k=2)
        assert [t.size for t in rep.topics] == [8, 2]

    def test_small_cluster_raises_gap_warning(self):
        from soup_cli.utils.topics import build_topic_report

        rows = self._rows("code", 99) + self._rows("safety", 1)
        rep = build_topic_report(
            rows, [0] * 99 + [1], k=2, min_fraction=0.02
        )
        assert any("thin" in w.lower() for w in rep.warnings)

    def test_no_warning_when_all_clusters_healthy(self):
        from soup_cli.utils.topics import build_topic_report

        rows = self._rows("code", 5) + self._rows("math", 5)
        rep = build_topic_report(rows, [0] * 5 + [1] * 5, k=2, min_fraction=0.02)
        assert rep.warnings == ()

    def test_member_indices_partition_the_input(self):
        from soup_cli.utils.topics import build_topic_report

        rows = self._rows("a", 3) + self._rows("b", 2)
        rep = build_topic_report(rows, [1, 0, 1, 0, 1], k=2)
        allidx = sorted(sum((list(t.member_indices) for t in rep.topics), []))
        assert allidx == [0, 1, 2, 3, 4]

    def test_zero_rows(self):
        from soup_cli.utils.topics import build_topic_report

        rep = build_topic_report([], [], k=1)
        assert rep.n_rows == 0
        assert rep.topics == ()

    def test_length_mismatch_rejected(self):
        from soup_cli.utils.topics import build_topic_report

        with pytest.raises(ValueError, match="same length"):
            build_topic_report(self._rows("a", 2), [0], k=1)

    def test_report_is_frozen(self):
        import dataclasses

        from soup_cli.utils.topics import build_topic_report

        rep = build_topic_report(self._rows("a", 2), [0, 0], k=1)
        with pytest.raises(dataclasses.FrozenInstanceError):
            rep.n_rows = 99


class TestDataTopicsCli:
    def _write(self, tmp_path, rows):
        path = tmp_path / "data.jsonl"
        path.write_text(
            "\n".join(json.dumps(r) for r in rows), encoding="utf-8"
        )
        return path

    def _rows(self):
        rows = [
            {"messages": [
                {"role": "user", "content": "q"},
                {"role": "assistant", "content": f"python function def {i}"},
            ]}
            for i in range(5)
        ]
        rows += [
            {"messages": [
                {"role": "user", "content": "q"},
                {"role": "assistant", "content": f"protein enzyme fold {i}"},
            ]}
            for i in range(5)
        ]
        return rows

    def _fake_vecs(self):
        import numpy as np

        return np.vstack([
            np.tile([1.0, 0.0], (5, 1)), np.tile([0.0, 1.0], (5, 1))
        ]).astype(np.float32)

    def test_help(self):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner().invoke(app, ["data", "topics", "--help"])
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert "cluster" in _clean(res.output).lower()

    def test_happy_path_writes_report(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_topics as cmd

        path = self._write(tmp_path, self._rows())
        monkeypatch.setattr(cmd, "embed_texts", lambda *a, **k: self._fake_vecs())
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app,
            ["data", "topics", str(path), "--clusters", "2", "-o", "topics.json"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        data = json.loads((tmp_path / "topics.json").read_text(encoding="utf-8"))
        assert data["n_rows"] == 10
        assert len(data["topics"]) == 2
        assert sum(t["fraction"] for t in data["topics"]) == pytest.approx(1.0)
        # the two synthetic domains must separate
        labels = " ".join(t["label"] for t in data["topics"])
        assert "python" in labels and "protein" in labels

    def test_auto_clusters_runs(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_topics as cmd

        path = self._write(tmp_path, self._rows())
        monkeypatch.setattr(cmd, "embed_texts", lambda *a, **k: self._fake_vecs())
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(app, ["data", "topics", str(path)])
        assert res.exit_code == 0, (res.output, repr(res.exception))

    def test_bad_clusters_value_rejected(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        path = self._write(tmp_path, self._rows())
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "topics", str(path), "--clusters", "banana"]
        )
        assert res.exit_code == 1
        assert "auto" in _clean(res.output).lower()

    def test_output_outside_cwd_rejected(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        path = self._write(tmp_path, self._rows())
        work = tmp_path / "work"
        work.mkdir()
        monkeypatch.chdir(work)
        res = CliRunner().invoke(
            app, ["data", "topics", str(path), "-o", str(tmp_path / "esc.json")]
        )
        assert res.exit_code == 1
        assert "outside" in _clean(res.output).lower()

    def test_missing_file(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(app, ["data", "topics", "nope.jsonl"])
        assert res.exit_code == 1
        assert "not found" in _clean(res.output).lower()

    def test_empty_dataset(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        path = tmp_path / "empty.jsonl"
        path.write_text("", encoding="utf-8")
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(app, ["data", "topics", str(path)])
        assert res.exit_code == 1
        assert "empty" in _clean(res.output).lower()

    def test_import_error_friendly(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_topics as cmd

        def _boom(*a, **k):
            raise ImportError("No module named 'torch'")

        monkeypatch.setattr(cmd, "embed_texts", _boom)
        path = self._write(tmp_path, self._rows())
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(app, ["data", "topics", str(path)])
        assert res.exit_code == 1
        assert "soup-cli[train]" in _clean(res.output)

    def test_pooling_refusal_surfaces(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_topics as cmd

        def _refuse(*a, **k):
            raise ValueError("cannot verify pooling for 'x/y'")

        monkeypatch.setattr(cmd, "embed_texts", _refuse)
        path = self._write(tmp_path, self._rows())
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(app, ["data", "topics", str(path)])
        assert res.exit_code == 1
        assert "pooling" in _clean(res.output)

    def test_markup_in_data_is_escaped(self, tmp_path, monkeypatch):
        """A crafted row must not inject Rich markup into the table."""
        import numpy as np
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_topics as cmd

        rows = [
            {"messages": [{"role": "assistant", "content": "[red]boom[/] alpha"}]}
        ] * 4
        path = self._write(tmp_path, rows)
        monkeypatch.setattr(
            cmd, "embed_texts",
            lambda *a, **k: np.tile([1.0, 0.0], (4, 1)).astype(np.float32),
        )
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "topics", str(path), "--clusters", "1"]
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))

    def test_gap_warning_surfaces_in_output(self, tmp_path, monkeypatch):
        import numpy as np
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_topics as cmd

        rows = [
            {"messages": [{"role": "assistant", "content": f"code {i}"}]}
            for i in range(99)
        ]
        rows += [{"messages": [{"role": "assistant", "content": "safety"}]}]
        path = self._write(tmp_path, rows)
        vecs = np.vstack([
            np.tile([1.0, 0.0], (99, 1)), np.tile([0.0, 1.0], (1, 1))
        ]).astype(np.float32)
        monkeypatch.setattr(cmd, "embed_texts", lambda *a, **k: vecs)
        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "topics", str(path), "--clusters", "2"]
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert "thin" in _clean(res.output).lower()


class TestComputePairLosses:
    """compute_eval_loss returns only the MEAN and silently compacts the
    list, so a skipped pair shifts every later index. Canary exposure needs
    per-item losses aligned to the input.
    """

    def _fake_torch_env(self, monkeypatch, loss_value=1.5):
        from soup_cli.utils import live_eval

        torch = pytest.importorskip("torch")

        def _fake_tokenize(tokenizer, prompt, target, *, max_length):
            if target == "":
                return torch.tensor([[1]]), torch.tensor([[-100]])
            return torch.tensor([[1, 2]]), torch.tensor([[-100, 2]])

        monkeypatch.setattr(live_eval, "_tokenize_pair", _fake_tokenize)

        class _Model:
            def eval(self):
                return self

            def __call__(self, **kwargs):
                class _Out:
                    loss = torch.tensor(loss_value)

                return _Out()

        return live_eval, _Model()

    def test_index_aligned_with_nan_for_skipped(self, monkeypatch):
        import math

        live_eval, model = self._fake_torch_env(monkeypatch)
        out = live_eval.compute_pair_losses(
            model, object(), [("p", "a"), ("p", ""), ("p", "b")], device="cpu"
        )
        assert len(out) == 3, "must be index-aligned with pairs"
        assert out[0] == pytest.approx(1.5)
        assert math.isnan(out[1]), "unusable pair must be nan, not dropped"
        assert out[2] == pytest.approx(1.5)

    def test_empty_pairs_returns_empty_list(self, monkeypatch):
        live_eval, model = self._fake_torch_env(monkeypatch)
        assert live_eval.compute_pair_losses(
            model, object(), [], device="cpu"
        ) == []

    def test_compute_eval_loss_is_mean_of_non_nan(self, monkeypatch):
        from soup_cli.utils import live_eval

        monkeypatch.setattr(
            live_eval, "compute_pair_losses",
            lambda *a, **k: [1.0, float("nan"), 3.0],
        )
        assert live_eval.compute_eval_loss(
            object(), object(), [("p", "a")] * 3, device="cpu"
        ) == pytest.approx(2.0)

    def test_compute_eval_loss_all_nan_returns_nan(self, monkeypatch):
        import math

        from soup_cli.utils import live_eval

        monkeypatch.setattr(
            live_eval, "compute_pair_losses", lambda *a, **k: [float("nan")]
        )
        assert math.isnan(
            live_eval.compute_eval_loss(
                object(), object(), [("p", "a")], device="cpu"
            )
        )

    def test_compute_eval_loss_empty_returns_nan(self, monkeypatch):
        import math

        from soup_cli.utils import live_eval

        monkeypatch.setattr(live_eval, "compute_pair_losses", lambda *a, **k: [])
        assert math.isnan(
            live_eval.compute_eval_loss(object(), object(), [], device="cpu")
        )

    def test_refactor_preserves_mean_behaviour(self, monkeypatch):
        """compute_eval_loss must equal mean(non-nan compute_pair_losses)."""
        live_eval, model = self._fake_torch_env(monkeypatch, loss_value=2.0)
        pairs = [("p", "a"), ("p", ""), ("p", "b")]
        losses = live_eval.compute_pair_losses(
            model, object(), pairs, device="cpu"
        )
        finite = [x for x in losses if x == x]
        assert live_eval.compute_eval_loss(
            model, object(), pairs, device="cpu"
        ) == pytest.approx(sum(finite) / len(finite))

    def test_max_length_still_validated(self, monkeypatch):
        live_eval, model = self._fake_torch_env(monkeypatch)
        with pytest.raises((ValueError, TypeError)):
            live_eval.compute_pair_losses(
                model, object(), [("p", "a")], device="cpu", max_length=0
            )


class TestCanaryGeneration:
    def test_deterministic_by_seed(self):
        from soup_cli.utils.canary import generate_canaries

        assert generate_canaries(count=5, seed=7) == generate_canaries(
            count=5, seed=7
        )

    def test_different_seeds_differ(self):
        from soup_cli.utils.canary import generate_canaries

        assert generate_canaries(count=5, seed=1) != generate_canaries(
            count=5, seed=2
        )

    def test_secrets_unique(self):
        from soup_cli.utils.canary import generate_canaries

        canaries = generate_canaries(count=50, seed=0)
        assert len({c.secret for c in canaries}) == 50

    def test_controls_exclude_inserted(self):
        from soup_cli.utils.canary import generate_canaries, generate_controls

        inserted = generate_canaries(count=10, seed=0)
        controls = generate_controls(
            count=20, seed=0, exclude={c.secret for c in inserted}
        )
        assert not ({c.secret for c in controls} & {c.secret for c in inserted})

    def test_controls_share_carrier_with_canaries(self):
        """Controls must vary ONLY the secret — else the comparison is
        measuring the carrier, not memorization."""
        from soup_cli.utils.canary import generate_canaries, generate_controls

        canary = generate_canaries(count=1, seed=0)[0]
        control = generate_controls(count=1, seed=99, exclude=set())[0]
        assert control.carrier == canary.carrier

    def test_secret_shape_is_from_the_declared_space(self):
        """Controls are only a valid null if they share the secret space."""
        import re

        from soup_cli.utils.canary import generate_canaries, generate_controls

        shape = re.compile(r"^ [0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}$")
        for canary in generate_canaries(count=8, seed=3):
            assert shape.match(canary.secret), canary.secret
        for control in generate_controls(count=8, seed=4, exclude=set()):
            assert shape.match(control.secret), control.secret

    @pytest.mark.parametrize("bad", [0, -1, True, "5", 10_001])
    def test_rejects_bad_count(self, bad):
        from soup_cli.utils.canary import generate_canaries

        with pytest.raises((ValueError, TypeError)):
            generate_canaries(count=bad, seed=0)

    def test_canary_rows_are_messages_format(self):
        from soup_cli.utils.canary import canary_rows, generate_canaries

        rows = canary_rows(generate_canaries(count=2, seed=0))
        assert len(rows) == 2
        for row in rows:
            assert set(row) == {"messages"}
            assert [m["role"] for m in row["messages"]] == ["user", "assistant"]

    def test_canary_is_frozen(self):
        import dataclasses

        from soup_cli.utils.canary import Canary

        canary = Canary(carrier="c", secret="s")
        with pytest.raises(dataclasses.FrozenInstanceError):
            canary.secret = "other"


class TestCanaryManifest:
    def test_roundtrip(self, tmp_path, monkeypatch):
        from soup_cli.utils.canary import (
            generate_canaries,
            load_manifest,
            write_manifest,
        )

        monkeypatch.chdir(tmp_path)
        canaries = generate_canaries(count=3, seed=0)
        write_manifest(canaries, "m.json")
        assert load_manifest("m.json") == canaries

    def test_outside_cwd_rejected(self, tmp_path, monkeypatch):
        from soup_cli.utils.canary import generate_canaries, write_manifest

        work = tmp_path / "work"
        work.mkdir()
        monkeypatch.chdir(work)
        with pytest.raises(ValueError):
            write_manifest(
                generate_canaries(count=1, seed=0), str(tmp_path / "esc.json")
            )

    def test_missing_manifest_is_a_friendly_message(self, tmp_path, monkeypatch):
        """Not a raw locale-dependent OS error.

        Live smoke surfaced "[WinError 2] Не удается найти указанный файл"
        leaking straight through from os.path.getsize.
        """
        from soup_cli.utils.canary import load_manifest

        monkeypatch.chdir(tmp_path)
        with pytest.raises(ValueError, match="manifest not found"):
            load_manifest("nope.json")

    def test_malformed_manifest_rejected(self, tmp_path, monkeypatch):
        from pathlib import Path

        from soup_cli.utils.canary import load_manifest

        monkeypatch.chdir(tmp_path)
        Path("bad.json").write_text("{not json", encoding="utf-8")
        with pytest.raises(ValueError, match="manifest"):
            load_manifest("bad.json")

    def test_non_list_canaries_rejected(self, tmp_path, monkeypatch):
        from pathlib import Path

        from soup_cli.utils.canary import load_manifest

        monkeypatch.chdir(tmp_path)
        Path("bad.json").write_text('{"canaries": "nope"}', encoding="utf-8")
        with pytest.raises(ValueError, match="manifest"):
            load_manifest("bad.json")

    def test_entry_missing_fields_rejected(self, tmp_path, monkeypatch):
        from pathlib import Path

        from soup_cli.utils.canary import load_manifest

        monkeypatch.chdir(tmp_path)
        Path("bad.json").write_text(
            '{"canaries": [{"carrier": "c"}]}', encoding="utf-8"
        )
        with pytest.raises(ValueError, match="carrier"):
            load_manifest("bad.json")

    def test_oversize_manifest_rejected(self, tmp_path, monkeypatch):
        from pathlib import Path

        from soup_cli.utils.canary import _MAX_MANIFEST_BYTES, load_manifest

        monkeypatch.chdir(tmp_path)
        Path("big.json").write_text(
            " " * (_MAX_MANIFEST_BYTES + 10), encoding="utf-8"
        )
        with pytest.raises(ValueError, match="too large"):
            load_manifest("big.json")

    @pytest.mark.skipif(
        __import__("os").name == "nt", reason="POSIX permissions only"
    )
    def test_manifest_is_not_world_readable(self, tmp_path, monkeypatch):
        """The manifest IS the secret. On a shared box a 0644 file lets any
        local user read every canary without ever running check."""
        import os
        import stat

        from soup_cli.utils.canary import generate_canaries, write_manifest

        monkeypatch.chdir(tmp_path)
        write_manifest(generate_canaries(count=2, seed=0), "m.json")
        mode = stat.S_IMODE(os.stat("m.json").st_mode)
        assert not (mode & stat.S_IRGRP), f"group-readable: {oct(mode)}"
        assert not (mode & stat.S_IROTH), f"world-readable: {oct(mode)}"

    def test_load_manifest_caps_entry_count(self, tmp_path, monkeypatch):
        """A manifest packed with entries would run unbounded forward passes."""
        from pathlib import Path

        from soup_cli.utils.canary import _MAX_CANARIES, load_manifest

        monkeypatch.chdir(tmp_path)
        entries = [
            {"carrier": "c", "secret": f"s{i}"}
            for i in range(_MAX_CANARIES + 1)
        ]
        Path("big.json").write_text(
            json.dumps({"canaries": entries}), encoding="utf-8"
        )
        with pytest.raises(ValueError, match="too many canaries"):
            load_manifest("big.json")

    @pytest.mark.skipif(
        not hasattr(__import__("os"), "symlink"), reason="POSIX only"
    )
    def test_symlinked_manifest_rejected(self, tmp_path, monkeypatch):
        import os

        from soup_cli.utils.canary import load_manifest

        monkeypatch.chdir(tmp_path)
        real = tmp_path / "real.json"
        real.write_text('{"canaries": []}', encoding="utf-8")
        try:
            os.symlink(real, tmp_path / "link.json")
        except (OSError, NotImplementedError):
            pytest.skip("symlink unavailable")
        with pytest.raises(ValueError):
            load_manifest("link.json")


class TestComputeExposure:
    """The release's core claim. Fixtures use INTEGER control losses so
    `cheaper / n_controls` is exact — a percentile boundary compared with
    `<=` must not hinge on float noise (the semdedup boundary test was
    vacuous for exactly that reason).
    """

    def _controls_0_to_99(self):
        return [float(i) for i in range(100)]

    def test_below_all_controls_is_memorized(self):
        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([0.1], [1.0, 2.0, 3.0, 4.0], ["s1"])
        assert out[0].percentile == 0.0
        assert out[0].memorized is True

    def test_middle_of_controls_not_memorized(self):
        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([2.5], [1.0, 2.0, 3.0, 4.0], ["s1"])
        assert out[0].percentile == pytest.approx(0.5)
        assert out[0].memorized is False

    def test_percentile_boundary_exactly_0_01_is_memorized(self):
        """1/100 is exact in binary — `<=` must include it."""
        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([0.5], self._controls_0_to_99(), ["s1"])
        assert out[0].percentile == 0.01, "fixture must land EXACTLY on 0.01"
        assert out[0].memorized is True, "<= 0.01 must count as memorized"

    def test_just_above_0_01_not_memorized(self):
        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([1.5], self._controls_0_to_99(), ["s1"])
        assert out[0].percentile == 0.02
        assert out[0].memorized is False

    def test_ties_count_as_not_strictly_less(self):
        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([1.0], [1.0, 1.0, 1.0, 1.0], ["s1"])
        assert out[0].percentile == 0.0

    def test_nan_canary_loss_is_unknown_not_memorized(self):
        """A NaN must never read as a leak."""
        import math

        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([float("nan")], [1.0, 2.0], ["s1"])
        assert out[0].memorized is False
        assert out[0].percentile == 1.0
        assert math.isnan(out[0].loss)

    def test_nan_controls_are_dropped_from_the_denominator(self):
        from soup_cli.utils.canary import compute_exposure

        out = compute_exposure([0.5], [1.0, float("nan"), 2.0], ["s1"])
        assert out[0].percentile == 0.0

    def test_zero_controls_refuses(self):
        from soup_cli.utils.canary import compute_exposure

        with pytest.raises(ValueError, match="at least one control"):
            compute_exposure([1.0], [], ["s1"])

    def test_all_nan_controls_refuses(self):
        """Refusing beats reporting OK against nothing."""
        from soup_cli.utils.canary import compute_exposure

        with pytest.raises(ValueError, match="at least one control"):
            compute_exposure([1.0], [float("nan")], ["s1"])

    def test_length_mismatch_rejected(self):
        from soup_cli.utils.canary import compute_exposure

        with pytest.raises(ValueError, match="same length"):
            compute_exposure([1.0, 2.0], [1.0], ["s1"])


class TestClassifyCanary:
    def _exposures(self, percentiles):
        """Build exposures landing on EXACT percentiles via 100 controls."""
        from soup_cli.utils.canary import compute_exposure

        controls = [float(i) for i in range(100)]
        losses = [pct * 100 - 0.5 for pct in percentiles]
        return compute_exposure(
            losses, controls, [f"s{i}" for i in range(len(losses))]
        )

    def test_every_canary_memorized_is_major(self):
        """The real signal: a memorized set lands at percentile 0.0."""
        from soup_cli.utils.canary import classify_canary

        assert classify_canary(self._exposures([0.0] * 10)) == "MAJOR"

    def test_two_of_ten_memorized_is_major(self):
        """A partial leak must still fire: P(X>=2 | K=10, p=.01) = 0.4%."""
        from soup_cli.utils.canary import classify_canary

        assert classify_canary(
            self._exposures([0.0, 0.01, 0.5, 0.6, 0.4, 0.7, 0.3, 0.8, 0.5, 0.9])
        ) == "MAJOR"

    def test_one_of_sixteen_memorized_is_not_major(self):
        """THE LIVE-SMOKE FIX. P(X>=1 | K=16, p=.01) = 15% — one canary
        dipping low is what a CLEAN model does one run in seven, and MAJOR
        exits 2. Firing here would make the CI gate cry wolf."""
        from soup_cli.utils.canary import classify_canary

        pcts = [0.0] + [0.3 + 0.04 * i for i in range(15)]
        assert classify_canary(self._exposures(pcts)) != "MAJOR"

    def test_the_observed_clean_model_is_ok(self):
        """Measured live: SmolLM2-135M that never saw the canaries.

        Two of ten under 10% — pure noise. The old any-canary rule called
        this MINOR; it must be OK.
        """
        from soup_cli.utils.canary import classify_canary

        observed = [0.609, 0.391, 0.930, 0.383, 0.328,
                    0.016, 0.609, 0.695, 0.156, 0.039]
        assert classify_canary(self._exposures(observed)) == "OK"

    def test_many_suspicious_is_minor(self):
        """5 of 10 under 10% is 100x chance — worth flagging, not MAJOR."""
        from soup_cli.utils.canary import classify_canary

        pcts = [0.02, 0.03, 0.05, 0.07, 0.09, 0.5, 0.6, 0.7, 0.8, 0.9]
        assert classify_canary(self._exposures(pcts)) == "MINOR"

    def test_all_typical_is_ok(self):
        from soup_cli.utils.canary import classify_canary

        assert classify_canary(self._exposures([0.5, 0.6])) == "OK"

    def test_empty_exposures_is_ok(self):
        from soup_cli.utils.canary import classify_canary

        assert classify_canary(()) == "OK"

    def test_single_canary_needs_percentile_zero_to_fire(self):
        from soup_cli.utils.canary import classify_canary

        assert classify_canary(self._exposures([0.0])) == "MAJOR"
        assert classify_canary(self._exposures([0.5])) == "OK"


class TestBinomialTail:
    def test_known_values(self):
        from soup_cli.utils.canary import _binomial_tail

        # P(X>=1 | n=16, p=.01) = 1 - .99^16
        assert _binomial_tail(1, 16, 0.01) == pytest.approx(1 - 0.99 ** 16)
        # P(X>=0) is certain; P(X>n) impossible
        assert _binomial_tail(0, 10, 0.1) == 1.0
        assert _binomial_tail(11, 10, 0.1) == 0.0

    def test_tail_shrinks_as_k_grows(self):
        from soup_cli.utils.canary import _binomial_tail

        tails = [_binomial_tail(k, 10, 0.01) for k in range(1, 5)]
        assert tails == sorted(tails, reverse=True)

    def test_false_alarm_rate_is_bounded_at_the_default_count(self):
        """The property the rule exists for: on a CLEAN model (percentiles
        ~Uniform), the chance of a MAJOR must sit under alpha — NOT the 15%
        the any-canary rule gave at K=16."""
        import random

        from soup_cli.utils.canary import CanaryExposure, classify_canary

        rng = random.Random(0)
        majors = 0
        trials = 2000
        for _ in range(trials):
            exposures = tuple(
                CanaryExposure(
                    secret=f"s{i}", loss=1.0, percentile=pct,
                    memorized=pct <= 0.01,
                )
                for i, pct in enumerate(rng.random() for _ in range(16))
            )
            if classify_canary(exposures) == "MAJOR":
                majors += 1
        rate = majors / trials
        assert rate < 0.05, f"false MAJOR rate {rate:.1%} on clean models"


class TestBuildCanaryReport:
    def test_assembles_report(self):
        from soup_cli.utils.canary import build_canary_report

        controls = [float(i) for i in range(100)]
        rep = build_canary_report([0.5], controls, ["s1"])
        assert rep.verdict == "MAJOR"
        assert rep.n_controls == 100
        assert len(rep.exposures) == 1

    def test_n_controls_excludes_nan(self):
        from soup_cli.utils.canary import build_canary_report

        rep = build_canary_report([5.0], [1.0, float("nan"), 2.0], ["s1"])
        assert rep.n_controls == 2

    def test_report_is_frozen(self):
        import dataclasses

        from soup_cli.utils.canary import build_canary_report

        rep = build_canary_report([5.0], [1.0, 2.0], ["s1"])
        with pytest.raises(dataclasses.FrozenInstanceError):
            rep.verdict = "OK"

    def test_to_dict_roundtrips_through_json(self):
        from soup_cli.utils.canary import (
            build_canary_report,
            canary_report_to_dict,
        )

        rep = build_canary_report([5.0], [1.0, 2.0], ["s1"])
        payload = canary_report_to_dict(rep)
        assert json.loads(json.dumps(payload)) == payload
        assert payload["verdict"] == "OK"
        assert payload["n_controls"] == 2

    def test_to_dict_nan_loss_becomes_null_not_zero(self):
        """0.0 would read as the strongest possible leak signal."""
        from soup_cli.utils.canary import (
            build_canary_report,
            canary_report_to_dict,
        )

        rep = build_canary_report([float("nan")], [1.0, 2.0], ["s1"])
        payload = canary_report_to_dict(rep)
        assert payload["exposures"][0]["loss"] is None


class TestDataCanaryCli:
    def _dataset(self, tmp_path, count=10):
        path = tmp_path / "d.jsonl"
        path.write_text(
            "\n".join(
                json.dumps({"messages": [
                    {"role": "user", "content": "q"},
                    {"role": "assistant", "content": "a"},
                ]})
                for _ in range(count)
            ),
            encoding="utf-8",
        )
        return path

    def test_insert_help(self):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner().invoke(app, ["data", "canary", "insert", "--help"])
        assert res.exit_code == 0, (res.output, repr(res.exception))

    def test_check_help(self):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner().invoke(app, ["data", "canary", "check", "--help"])
        assert res.exit_code == 0, (res.output, repr(res.exception))

    def test_insert_adds_k_rows_and_manifest(self, tmp_path, monkeypatch):
        from pathlib import Path

        from typer.testing import CliRunner

        from soup_cli.cli import app

        monkeypatch.chdir(tmp_path)
        src = self._dataset(tmp_path, count=10)
        res = CliRunner().invoke(
            app,
            ["data", "canary", "insert", str(src), "-o", "out.jsonl",
             "--count", "4", "--manifest", "m.json"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        rows = [
            json.loads(x)
            for x in Path("out.jsonl").read_text().splitlines() if x
        ]
        assert len(rows) == 14
        manifest = json.loads(Path("m.json").read_text())
        assert len(manifest["canaries"]) == 4

    def test_inserted_rows_carry_the_manifest_secrets(self, tmp_path, monkeypatch):
        """The manifest must describe what actually landed in the data."""
        from pathlib import Path

        from typer.testing import CliRunner

        from soup_cli.cli import app

        monkeypatch.chdir(tmp_path)
        src = self._dataset(tmp_path, count=5)
        res = CliRunner().invoke(
            app,
            ["data", "canary", "insert", str(src), "-o", "out.jsonl",
             "--count", "3", "--manifest", "m.json"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        blob = Path("out.jsonl").read_text()
        for entry in json.loads(Path("m.json").read_text())["canaries"]:
            assert entry["secret"].strip() in blob

    def test_insert_warns_manifest_is_sensitive(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        monkeypatch.chdir(tmp_path)
        src = self._dataset(tmp_path, count=2)
        res = CliRunner().invoke(
            app, ["data", "canary", "insert", str(src), "-o", "o.jsonl",
                  "--manifest", "m.json"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        out = _clean(res.output).lower()
        assert "commit" in out or "secret" in out

    def test_manifest_is_written_before_the_poisoned_dataset(
        self, tmp_path, monkeypatch
    ):
        """Order matters: a dataset with no manifest is unauditable.

        If the data were written first and the manifest then failed, a
        canary-poisoned dataset would survive on disk with nothing left to
        say what was inserted into it.
        """
        from pathlib import Path

        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        src = self._dataset(tmp_path, count=3)

        def _boom(canaries, path):
            raise OSError("disk full")

        monkeypatch.setattr(cmd, "write_manifest", _boom)
        res = CliRunner().invoke(
            app,
            ["data", "canary", "insert", str(src), "-o", "out.jsonl",
             "--manifest", "m.json"],
        )
        assert res.exit_code == 1
        assert not Path("out.jsonl").exists(), (
            "a canary-poisoned dataset must not survive a manifest failure"
        )

    def test_data_write_failure_warns_the_manifest_is_now_stale(
        self, tmp_path, monkeypatch
    ):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        src = self._dataset(tmp_path, count=3)

        def _boom(text, path):
            raise OSError("disk full")

        monkeypatch.setattr(cmd, "atomic_write_text", _boom)
        res = CliRunner().invoke(
            app,
            ["data", "canary", "insert", str(src), "-o", "out.jsonl",
             "--manifest", "m.json"],
        )
        assert res.exit_code == 1
        out = _clean(res.output).lower()
        assert "not inserted" in out or "re-run" in out

    def test_insert_output_outside_cwd_rejected(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        src = self._dataset(tmp_path, count=2)
        work = tmp_path / "work"
        work.mkdir()
        monkeypatch.chdir(work)
        res = CliRunner().invoke(
            app,
            ["data", "canary", "insert", str(src),
             "-o", str(tmp_path / "esc.jsonl"), "--manifest", "m.json"],
        )
        assert res.exit_code == 1
        assert "outside" in _clean(res.output).lower()

    def test_insert_manifest_outside_cwd_rejected(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        src = self._dataset(tmp_path, count=2)
        work = tmp_path / "work"
        work.mkdir()
        monkeypatch.chdir(work)
        res = CliRunner().invoke(
            app,
            ["data", "canary", "insert", str(src), "-o", "o.jsonl",
             "--manifest", str(tmp_path / "esc.json")],
        )
        assert res.exit_code == 1
        assert "outside" in _clean(res.output).lower()

    def _seed_manifest(self, tmp_path, count=2):
        from soup_cli.utils.canary import generate_canaries, write_manifest

        write_manifest(generate_canaries(count=count, seed=0), "m.json")

    def test_check_major_exits_2(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        self._seed_manifest(tmp_path, count=2)
        monkeypatch.setattr(
            cmd, "_load_pair", lambda *a, **k: (object(), object(), "cpu")
        )

        def _losses(model, tok, pairs, **kwargs):
            # first 2 are the canaries -> far cheaper than every control
            return [0.01 if i < 2 else float(i) for i in range(len(pairs))]

        monkeypatch.setattr(cmd, "compute_pair_losses", _losses)
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "fake/model", "--controls", "16"],
        )
        assert res.exit_code == 2, (res.output, repr(res.exception))
        assert "MAJOR" in _clean(res.output)

    def test_check_ok_exits_0(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        self._seed_manifest(tmp_path, count=2)
        monkeypatch.setattr(
            cmd, "_load_pair", lambda *a, **k: (object(), object(), "cpu")
        )
        # Canaries sit mid-distribution: 5 of 16 controls are cheaper ->
        # percentile 0.31 -> OK. (An all-equal fixture would make NO control
        # strictly cheaper -> percentile 0.0 -> MAJOR, which is the rule
        # working correctly on an unrealistic input.)
        monkeypatch.setattr(
            cmd, "compute_pair_losses",
            lambda model, tok, pairs, **kw: (
                [5.0] * 2 + [float(i) for i in range(len(pairs) - 2)]
            ),
        )
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "fake/model", "--controls", "16"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert "OK" in _clean(res.output)

    def test_check_splits_canary_and_control_losses_correctly(
        self, tmp_path, monkeypatch
    ):
        """The canary/control split must follow the manifest length."""
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        self._seed_manifest(tmp_path, count=3)
        monkeypatch.setattr(
            cmd, "_load_pair", lambda *a, **k: (object(), object(), "cpu")
        )
        seen = {}

        def _losses(model, tok, pairs, **kwargs):
            seen["n_pairs"] = len(pairs)
            return [5.0] * 3 + [float(i) for i in range(len(pairs) - 3)]

        monkeypatch.setattr(cmd, "compute_pair_losses", _losses)
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "fake/model", "--controls", "8", "-o", "r.json"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert seen["n_pairs"] == 3 + 8, "3 canaries + 8 controls"
        from pathlib import Path

        report = json.loads(Path("r.json").read_text())
        assert len(report["exposures"]) == 3
        assert report["n_controls"] == 8

    def test_check_writes_report(self, tmp_path, monkeypatch):
        from pathlib import Path

        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        self._seed_manifest(tmp_path, count=1)
        monkeypatch.setattr(
            cmd, "_load_pair", lambda *a, **k: (object(), object(), "cpu")
        )
        monkeypatch.setattr(
            cmd, "compute_pair_losses",
            lambda model, tok, pairs, **kw: (
                [5.0] + [float(i) for i in range(len(pairs) - 1)]
            ),
        )
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "fake/model", "-o", "r.json", "--controls", "8"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        report = json.loads(Path("r.json").read_text())
        assert report["verdict"] == "OK"
        assert report["n_controls"] == 8

    def test_check_model_load_import_error_is_friendly(
        self, tmp_path, monkeypatch
    ):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        self._seed_manifest(tmp_path, count=1)

        def _boom(*a, **k):
            raise ImportError("No module named 'torch'")

        monkeypatch.setattr(cmd, "_load_pair", _boom)
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "fake/model"],
        )
        assert res.exit_code == 1
        assert "soup-cli[train]" in _clean(res.output)

    def test_check_model_load_failure_is_friendly(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        self._seed_manifest(tmp_path, count=1)

        def _boom(*a, **k):
            raise OSError("no such model")

        monkeypatch.setattr(cmd, "_load_pair", _boom)
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "nope/model"],
        )
        assert res.exit_code == 1
        assert "could not load" in _clean(res.output).lower()

    def test_esc_bytes_from_a_manifest_are_stripped(self, tmp_path, monkeypatch):
        """rich escape() neutralises [...] but NOT raw ESC bytes.

        The manifest is a shareable artifact ("anyone holding it can
        reproduce the canaries"), so a hostile one is a realistic input. An
        OSC 52 sequence in a secret would reach the terminal raw and could
        obscure the MAJOR verdict printed right below the table.
        """
        from pathlib import Path

        from typer.testing import CliRunner

        from soup_cli.cli import app
        from soup_cli.commands import data_canary as cmd

        monkeypatch.chdir(tmp_path)
        evil = "7c3f\x1b]52;c;ZXZpbA==\x07-9a21"
        Path("m.json").write_text(
            json.dumps({"canaries": [{"carrier": "c", "secret": evil}]}),
            encoding="utf-8",
        )
        monkeypatch.setattr(
            cmd, "_load_pair", lambda *a, **k: (object(), object(), "cpu")
        )
        monkeypatch.setattr(
            cmd, "compute_pair_losses",
            lambda model, tok, pairs, **kw: (
                [5.0] + [float(i) for i in range(len(pairs) - 1)]
            ),
        )
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "fake/model", "--controls", "8"],
        )
        assert res.exit_code == 0, (res.output, repr(res.exception))
        assert "\x1b" not in res.output, (
            "a raw ESC byte from the manifest reached the terminal"
        )

    def test_check_missing_manifest(self, tmp_path, monkeypatch):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        monkeypatch.chdir(tmp_path)
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "nope.json",
                  "--base", "m"],
        )
        assert res.exit_code == 1

    def test_check_empty_manifest_refuses(self, tmp_path, monkeypatch):
        from pathlib import Path

        from typer.testing import CliRunner

        from soup_cli.cli import app

        monkeypatch.chdir(tmp_path)
        Path("m.json").write_text('{"canaries": []}', encoding="utf-8")
        res = CliRunner().invoke(
            app, ["data", "canary", "check", "--manifest", "m.json",
                  "--base", "m"],
        )
        assert res.exit_code == 1
        assert "no canaries" in _clean(res.output).lower()


_REPLAY_YAML = """
base: HuggingFaceTB/SmolLM2-135M-Instruct
task: sft
data:
  train: train.jsonl
  replay: old.jsonl
  replay_ratio: 0.2
training:
  epochs: 1
"""

_PLAIN_YAML = (
    "base: m\ntask: sft\ndata:\n  train: t.jsonl\ntraining:\n  epochs: 1\n"
)


class TestReplaySchema:
    def _load(self, yaml_str):
        from soup_cli.config.loader import load_config_from_string

        return load_config_from_string(yaml_str)

    def test_happy_path(self):
        cfg = self._load(_REPLAY_YAML)
        assert cfg.data.replay == "old.jsonl"
        assert cfg.data.replay_ratio == 0.2

    def test_default_is_off(self):
        cfg = self._load(_PLAIN_YAML)
        assert cfg.data.replay is None
        assert cfg.data.replay_ratio == 0.1
        assert cfg.data.replay_seed is None

    def test_pretrain_allowed(self):
        cfg = self._load(
            _REPLAY_YAML.replace("task: sft", "task: pretrain").replace(
                "  train: train.jsonl",
                "  train: train.jsonl\n  format: plaintext",
            )
        )
        assert cfg.data.replay == "old.jsonl"

    def test_rejected_on_dpo(self):
        with pytest.raises(Exception, match="replay"):
            self._load(
                _REPLAY_YAML.replace("task: sft", "task: dpo").replace(
                    "  train: train.jsonl",
                    "  train: train.jsonl\n  format: dpo",
                )
            )

    def test_footgun_ratio_without_replay(self):
        yaml_str = (
            "base: m\ntask: sft\ndata:\n  train: t.jsonl\n"
            "  replay_ratio: 0.3\ntraining:\n  epochs: 1\n"
        )
        with pytest.raises(Exception, match="data.replay"):
            self._load(yaml_str)

    def test_footgun_seed_without_replay(self):
        yaml_str = (
            "base: m\ntask: sft\ndata:\n  train: t.jsonl\n"
            "  replay_seed: 7\ntraining:\n  epochs: 1\n"
        )
        with pytest.raises(Exception, match="data.replay"):
            self._load(yaml_str)

    def test_mutually_exclusive_with_packing(self):
        yaml_str = _REPLAY_YAML.replace(
            "training:\n  epochs: 1", "training:\n  epochs: 1\n  packing: true"
        )
        with pytest.raises(Exception, match="packing"):
            self._load(yaml_str)

    def test_mutually_exclusive_with_multipack(self):
        yaml_str = _REPLAY_YAML.replace(
            "training:\n  epochs: 1",
            "training:\n  epochs: 1\n  multipack: true",
        )
        with pytest.raises(Exception, match="multipack"):
            self._load(yaml_str)

    @pytest.mark.parametrize("bad", ["0.0", "0.6", "1.0", "-0.1"])
    def test_ratio_bounds(self, bad):
        with pytest.raises(Exception):
            self._load(
                _REPLAY_YAML.replace("replay_ratio: 0.2", f"replay_ratio: {bad}")
            )

    def test_ratio_boundary_0_5_allowed(self):
        cfg = self._load(
            _REPLAY_YAML.replace("replay_ratio: 0.2", "replay_ratio: 0.5")
        )
        assert cfg.data.replay_ratio == 0.5

    @pytest.mark.parametrize("bad", ['""', '"   "'])
    def test_replay_field_validator_rejects_blank(self, bad):
        with pytest.raises(Exception):
            self._load(_REPLAY_YAML.replace("replay: old.jsonl", f"replay: {bad}"))

    def test_replay_rejects_null_byte(self):
        from soup_cli.config.schema import DataConfig

        with pytest.raises(Exception, match="null"):
            DataConfig(train="t.jsonl", replay="a\x00b")

    def test_replay_rejects_overlong_path(self):
        from soup_cli.config.schema import DataConfig

        with pytest.raises(Exception, match="too long"):
            DataConfig(train="t.jsonl", replay="x" * 5000)

    def test_ratio_rejects_bool(self):
        with pytest.raises(Exception):
            self._load(
                _REPLAY_YAML.replace("replay_ratio: 0.2", "replay_ratio: true")
            )

    def test_seed_rejects_bool(self):
        with pytest.raises(Exception):
            self._load(
                _REPLAY_YAML.replace(
                    "replay_ratio: 0.2", "replay_seed: true"
                )
            )

    def test_replay_survives_model_dump(self):
        """Provenance rides the schema — the tracker/registry capture it."""
        cfg = self._load(_REPLAY_YAML)
        dumped = cfg.model_dump()
        assert dumped["data"]["replay"] == "old.jsonl"
        assert dumped["data"]["replay_ratio"] == 0.2


class TestResolveReplayCount:
    def test_the_locked_example(self):
        from soup_cli.utils.rehearsal import resolve_replay_count

        assert resolve_replay_count(1000, 0.1) == 111

    def test_final_fraction_really_is_the_ratio(self):
        """The whole point: r is a share of the FINAL set, not of the new."""
        from soup_cli.utils.rehearsal import resolve_replay_count

        n_replay = resolve_replay_count(1000, 0.1)
        assert n_replay / (1000 + n_replay) == pytest.approx(0.1, abs=1e-3)

    def test_naive_reading_is_not_used(self):
        """r * n_new = 100 would give 9.09% — the wrong reading."""
        from soup_cli.utils.rehearsal import resolve_replay_count

        assert resolve_replay_count(1000, 0.1) != 100

    def test_ratio_half_doubles_the_set(self):
        from soup_cli.utils.rehearsal import resolve_replay_count

        assert resolve_replay_count(100, 0.5) == 100

    def test_zero_new_rows(self):
        from soup_cli.utils.rehearsal import resolve_replay_count

        assert resolve_replay_count(0, 0.1) == 0

    @pytest.mark.parametrize("bad", [0.0, 0.9, 1.0, -0.1, True, "x"])
    def test_rejects_bad_ratio(self, bad):
        from soup_cli.utils.rehearsal import resolve_replay_count

        with pytest.raises((ValueError, TypeError)):
            resolve_replay_count(100, bad)

    @pytest.mark.parametrize("bad", [-1, True, 1.5, "10"])
    def test_rejects_bad_n_new(self, bad):
        from soup_cli.utils.rehearsal import resolve_replay_count

        with pytest.raises((ValueError, TypeError)):
            resolve_replay_count(bad, 0.1)


class TestMixReplay:
    def _rows(self, tag, count):
        return [
            {"messages": [{"role": "assistant", "content": f"{tag}{i}"}]}
            for i in range(count)
        ]

    def _tags(self, mixed):
        return [row["messages"][0]["content"] for row in mixed]

    def test_counts(self):
        from soup_cli.utils.rehearsal import mix_replay

        mixed, rep = mix_replay(
            self._rows("new", 1000), self._rows("old", 500), ratio=0.1, seed=0
        )
        assert rep.n_new == 1000
        assert rep.n_replay == 111
        assert rep.n_final == 1111
        assert len(mixed) == 1111
        assert rep.shortfall == 0

    def test_replay_rows_are_interleaved_not_appended(self):
        """The correctness crux: a trailing block is not rehearsal.

        `new + replay` puts every replay row in the final steps -- a second
        mini-finetune, which is what replay exists to avoid.
        """
        from soup_cli.utils.rehearsal import mix_replay

        mixed, rep = mix_replay(
            self._rows("new", 900), self._rows("old", 500), ratio=0.1, seed=0
        )
        positions = [
            i for i, tag in enumerate(self._tags(mixed)) if tag.startswith("old")
        ]
        assert len(positions) == rep.n_replay
        first_half = sum(1 for p in positions if p < len(mixed) / 2)
        assert first_half > 0, "replay rows must appear in the first half"
        assert first_half < len(positions), "and in the second half"

    def test_replay_is_not_clustered_at_the_tail(self):
        """Sharper than the halves check: the mean position of replay rows
        should sit near the middle, not near the end."""
        from soup_cli.utils.rehearsal import mix_replay

        mixed, _ = mix_replay(
            self._rows("new", 900), self._rows("old", 500), ratio=0.1, seed=0
        )
        tags = self._tags(mixed)
        positions = [i for i, t in enumerate(tags) if t.startswith("old")]
        mean_pos = sum(positions) / len(positions) / len(tags)
        assert 0.35 < mean_pos < 0.65, (
            f"replay rows cluster at {mean_pos:.2f} of the run — not interleaved"
        )

    def test_undersized_replay_uses_all_and_reports_shortfall(self):
        from soup_cli.utils.rehearsal import mix_replay

        mixed, rep = mix_replay(
            self._rows("new", 1000), self._rows("old", 10), ratio=0.1, seed=0
        )
        assert rep.requested == 111
        assert rep.n_replay == 10
        assert rep.shortfall == 101
        assert len(mixed) == 1010

    def test_no_upsampling_on_shortfall(self):
        """Repeating rows would silently change epoch semantics."""
        from soup_cli.utils.rehearsal import mix_replay

        mixed, _ = mix_replay(
            self._rows("new", 1000), self._rows("old", 10), ratio=0.1, seed=0
        )
        old = [t for t in self._tags(mixed) if t.startswith("old")]
        assert len(old) == len(set(old))

    def test_sampling_without_replacement(self):
        from soup_cli.utils.rehearsal import mix_replay

        mixed, _ = mix_replay(
            self._rows("new", 90), self._rows("old", 500), ratio=0.1, seed=0
        )
        old = [t for t in self._tags(mixed) if t.startswith("old")]
        assert len(old) == len(set(old))

    def test_every_new_row_survives(self):
        """Replay must ADD rehearsal, never displace the new task."""
        from soup_cli.utils.rehearsal import mix_replay

        new = self._rows("new", 100)
        mixed, _ = mix_replay(new, self._rows("old", 100), ratio=0.1, seed=0)
        kept = {t for t in self._tags(mixed) if t.startswith("new")}
        assert kept == {f"new{i}" for i in range(100)}

    def test_deterministic_by_seed(self):
        from soup_cli.utils.rehearsal import mix_replay

        args = (self._rows("new", 100), self._rows("old", 100))
        first, _ = mix_replay(*args, ratio=0.1, seed=42)
        second, _ = mix_replay(*args, ratio=0.1, seed=42)
        assert first == second

    def test_different_seeds_differ(self):
        from soup_cli.utils.rehearsal import mix_replay

        args = (self._rows("new", 100), self._rows("old", 100))
        first, _ = mix_replay(*args, ratio=0.1, seed=1)
        second, _ = mix_replay(*args, ratio=0.1, seed=2)
        assert first != second

    def test_none_seed_is_deterministic(self):
        """seed=None must not mean 'random' — runs must reproduce."""
        from soup_cli.utils.rehearsal import mix_replay

        args = (self._rows("new", 100), self._rows("old", 100))
        first, _ = mix_replay(*args, ratio=0.1, seed=None)
        second, _ = mix_replay(*args, ratio=0.1, seed=None)
        assert first == second

    def test_empty_replay_is_noop(self):
        from soup_cli.utils.rehearsal import mix_replay

        new = self._rows("new", 10)
        mixed, rep = mix_replay(new, [], ratio=0.1, seed=0)
        assert mixed == new
        assert rep.n_replay == 0
        assert rep.ratio_actual == 0.0

    def test_ratio_actual_reported(self):
        from soup_cli.utils.rehearsal import mix_replay

        _, rep = mix_replay(
            self._rows("new", 1000), self._rows("old", 500), ratio=0.1, seed=0
        )
        assert rep.ratio_actual == pytest.approx(111 / 1111)

    def test_report_is_frozen(self):
        import dataclasses

        from soup_cli.utils.rehearsal import mix_replay

        _, rep = mix_replay(
            self._rows("new", 10), self._rows("old", 10), ratio=0.1, seed=0
        )
        with pytest.raises(dataclasses.FrozenInstanceError):
            rep.n_replay = 99


class TestLoaderReplay:
    def _write(self, path, rows):
        path.write_text(
            "\n".join(json.dumps(r) for r in rows), encoding="utf-8"
        )

    def _chat(self, tag, count):
        return [
            {"messages": [
                {"role": "user", "content": "q"},
                {"role": "assistant", "content": f"{tag}{i}"},
            ]}
            for i in range(count)
        ]

    def _answers(self, rows):
        return [r["messages"][-1]["content"] for r in rows]

    def test_replay_mixed_into_train(self, tmp_path, monkeypatch):
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        self._write(tmp_path / "new.jsonl", self._chat("new", 90))
        self._write(tmp_path / "old.jsonl", self._chat("old", 90))
        cfg = DataConfig(
            train="new.jsonl", replay="old.jsonl",
            replay_ratio=0.1, replay_seed=0, val_split=0.0,
        )
        out = load_dataset(cfg)
        answers = self._answers(out["train"])
        assert sum(1 for a in answers if a.startswith("old")) == 10
        assert len(out["train"]) == 100

    def test_val_stays_pure_new_task(self, tmp_path, monkeypatch):
        """Replay must NOT leak into val — it is the new-task yardstick."""
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        self._write(tmp_path / "new.jsonl", self._chat("new", 100))
        self._write(tmp_path / "old.jsonl", self._chat("old", 100))
        cfg = DataConfig(
            train="new.jsonl", replay="old.jsonl",
            replay_ratio=0.2, replay_seed=0, val_split=0.2,
        )
        out = load_dataset(cfg)
        assert out["val"], "val must be non-empty for this test to mean anything"
        assert all(
            a.startswith("new") for a in self._answers(out["val"])
        ), "val must contain no replay rows"

    def test_no_replay_is_unchanged(self, tmp_path, monkeypatch):
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        self._write(tmp_path / "new.jsonl", self._chat("new", 10))
        cfg = DataConfig(train="new.jsonl", val_split=0.0)
        out = load_dataset(cfg)
        assert len(out["train"]) == 10
        assert self._answers(out["train"]) == [f"new{i}" for i in range(10)]

    def test_replay_file_gets_its_own_format_detection(
        self, tmp_path, monkeypatch
    ):
        """New data is chat, old data is alpaca -> both must normalize."""
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        self._write(tmp_path / "new.jsonl", self._chat("new", 90))
        self._write(
            tmp_path / "old.jsonl",
            [
                {"instruction": f"old-q{i}", "output": f"old{i}"}
                for i in range(90)
            ],
        )
        cfg = DataConfig(
            train="new.jsonl", replay="old.jsonl",
            replay_ratio=0.1, replay_seed=0, val_split=0.0,
        )
        out = load_dataset(cfg)
        assert len(out["train"]) == 100
        for row in out["train"]:
            assert "messages" in row, (
                "the alpaca replay file must be normalized via its OWN format "
                "detection, not the new file's"
            )
        assert sum(
            1 for a in self._answers(out["train"]) if a.startswith("old")
        ) == 10

    def test_missing_replay_file_raises(self, tmp_path, monkeypatch):
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        self._write(tmp_path / "new.jsonl", self._chat("new", 10))
        cfg = DataConfig(train="new.jsonl", replay="nope.jsonl", val_split=0.0)
        with pytest.raises((FileNotFoundError, ValueError), match="replay"):
            load_dataset(cfg)

    def test_replay_outside_cwd_rejected(self, tmp_path, monkeypatch):
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        work = tmp_path / "work"
        work.mkdir()
        self._write(work / "new.jsonl", self._chat("new", 10))
        self._write(tmp_path / "old.jsonl", self._chat("old", 10))
        monkeypatch.chdir(work)
        cfg = DataConfig(
            train="new.jsonl", replay=str(tmp_path / "old.jsonl"),
            val_split=0.0,
        )
        with pytest.raises(ValueError, match="outside"):
            load_dataset(cfg)

    def test_shortfall_is_reported_not_upsampled(self, tmp_path, monkeypatch):
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        self._write(tmp_path / "new.jsonl", self._chat("new", 90))
        self._write(tmp_path / "old.jsonl", self._chat("old", 3))
        cfg = DataConfig(
            train="new.jsonl", replay="old.jsonl",
            replay_ratio=0.1, replay_seed=0, val_split=0.0,
        )
        out = load_dataset(cfg)
        old = [a for a in self._answers(out["train"]) if a.startswith("old")]
        assert len(old) == 3
        assert len(set(old)) == 3, "rows must not be repeated"

    def _llava(self, tag, image, count):
        return [
            {
                "image": image,
                "conversations": [
                    {"from": "human", "value": "<image>\ndescribe"},
                    {"from": "gpt", "value": f"{tag}{i}"},
                ],
            }
            for i in range(count)
        ]

    def test_replay_vision_rows_get_traversal_protection(
        self, tmp_path, monkeypatch
    ):
        """The replay file must get the SAME image containment as the main one.

        load_dataset runs _validate_vision_images on the primary dataset
        (it exists to reject `{"image": "/etc/passwd"}`), but the replay
        file is loaded by its own path. A genuinely llava-shaped replay row
        keeps its `image` value through format_to_messages, so without a
        validation pass a traversal path would reach PIL.Image.open.
        """
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        (tmp_path / "ok.png").write_bytes(b"\x89PNG\r\n\x1a\n")
        self._write(tmp_path / "new.jsonl", self._llava("new", "ok.png", 90))
        self._write(
            tmp_path / "old.jsonl",
            self._llava("old", "../../../../../../etc/passwd", 90),
        )
        cfg = DataConfig(
            train="new.jsonl", format="llava", replay="old.jsonl",
            replay_ratio=0.1, replay_seed=0, val_split=0.0,
        )
        out = load_dataset(cfg)
        escaped = [
            row for row in out["train"]
            if "etc/passwd" in str(row.get("image", ""))
        ]
        assert not escaped, (
            "a traversal image path from the replay file reached the mix "
            "unvalidated"
        )

    def test_replay_audio_rows_get_traversal_protection(
        self, tmp_path, monkeypatch
    ):
        from soup_cli.config.schema import DataConfig
        from soup_cli.data.loader import load_dataset

        monkeypatch.chdir(tmp_path)
        (tmp_path / "ok.wav").write_bytes(b"RIFF....WAVE")
        self._write(
            tmp_path / "new.jsonl",
            [{"audio": "ok.wav", "text": f"new{i}"} for i in range(90)],
        )
        self._write(
            tmp_path / "old.jsonl",
            [
                {"audio": "../../../../../../etc/shadow", "text": f"old{i}"}
                for i in range(90)
            ],
        )
        cfg = DataConfig(
            train="new.jsonl", format="asr", replay="old.jsonl",
            replay_ratio=0.1, replay_seed=0, val_split=0.0,
        )
        out = load_dataset(cfg)
        escaped = [
            row for row in out["train"]
            if "etc/shadow" in str(row.get("audio", ""))
        ]
        assert not escaped, (
            "a traversal audio path from the replay file reached the mix "
            "unvalidated"
        )

    def test_finalize_is_the_single_seam(self):
        """All three load paths must route through _finalize, or replay
        would silently apply to some datasets and not others."""
        import inspect

        from soup_cli.data import loader

        source = inspect.getsource(loader)
        assert source.count("def _finalize(") == 1
        # local + remote + hf paths
        assert source.count("_finalize(") >= 4


class TestTrainReplayFlags:
    def test_help_lists_replay(self):
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner(env={"COLUMNS": "200"}).invoke(app, ["train", "--help"])
        assert res.exit_code == 0, (res.output, repr(res.exception))
        cleaned = _clean(res.output)
        assert "--replay" in cleaned
        assert "--replay-ratio" in cleaned
        assert "--replay-seed" in cleaned

    def test_seed_override(self):
        from soup_cli.commands.train import _apply_replay_overrides

        out = _apply_replay_overrides(
            self._cfg(_REPLAY_YAML), replay=None, replay_ratio=None,
            replay_seed=7,
        )
        assert out.data.replay_seed == 7

    def test_seed_without_replay_rejected(self):
        from soup_cli.commands.train import _apply_replay_overrides

        with pytest.raises(Exception, match="replay"):
            _apply_replay_overrides(
                self._cfg(), replay=None, replay_ratio=None, replay_seed=7
            )

    def _cfg(self, yaml_str=None):
        from soup_cli.config.loader import load_config_from_string

        return load_config_from_string(yaml_str or _PLAIN_YAML)

    def test_apply_replay_overrides(self):
        from soup_cli.commands.train import _apply_replay_overrides

        out = _apply_replay_overrides(
            self._cfg(), replay="old.jsonl", replay_ratio=0.25
        )
        assert out.data.replay == "old.jsonl"
        assert out.data.replay_ratio == 0.25

    def test_no_flags_is_identity(self):
        from soup_cli.commands.train import _apply_replay_overrides

        cfg = self._cfg()
        out = _apply_replay_overrides(cfg, replay=None, replay_ratio=None)
        assert out is cfg
        assert out.data.replay is None

    def test_override_is_revalidated_against_the_gate(self):
        """--replay on task=dpo must hit _validate_replay_compat.

        A CLI override that skipped re-validation would slip past every
        cross-validator that YAML has to satisfy.
        """
        from soup_cli.commands.train import _apply_replay_overrides

        dpo = self._cfg(
            "base: m\ntask: dpo\ndata:\n  train: t.jsonl\n  format: dpo\n"
            "training:\n  epochs: 1\n"
        )
        with pytest.raises(Exception, match="replay"):
            _apply_replay_overrides(dpo, replay="old.jsonl", replay_ratio=None)

    def test_ratio_without_replay_flag_rejected(self):
        from soup_cli.commands.train import _apply_replay_overrides

        with pytest.raises(Exception, match="replay"):
            _apply_replay_overrides(self._cfg(), replay=None, replay_ratio=0.3)

    def test_bad_ratio_rejected(self):
        from soup_cli.commands.train import _apply_replay_overrides

        with pytest.raises(Exception):
            _apply_replay_overrides(
                self._cfg(), replay="old.jsonl", replay_ratio=0.9
            )

    def test_yaml_value_preserved_when_flag_absent(self):
        """--replay-ratio alone must not clobber a YAML data.replay."""
        from soup_cli.commands.train import _apply_replay_overrides

        cfg = self._cfg(_REPLAY_YAML)
        out = _apply_replay_overrides(cfg, replay=None, replay_ratio=0.3)
        assert out.data.replay == "old.jsonl"
        assert out.data.replay_ratio == 0.3

    def test_override_does_not_mutate_the_input_config(self):
        from soup_cli.commands.train import _apply_replay_overrides

        cfg = self._cfg()
        _apply_replay_overrides(cfg, replay="old.jsonl", replay_ratio=0.2)
        assert cfg.data.replay is None, "input config must not be mutated"


class TestLocalModelSizeGate:
    """The v0.71.36 replay smoke found this: `soup merge` writes a dir like
    ./denseA whose NAME carries no size marker, so model_size_from_name fell
    through to the 7B default, the hardware-fit gate predicted 16.3 GB and
    REFUSED to train a 135M model — blocking merge -> train-from-merged, the
    ordinary continual-learning flow --replay exists for. Same class as the
    v0.71.32 whisper and v0.71.33 "M suffix" fixes.
    """

    def _fake_checkpoint(self, tmp_path, shapes):
        """Write a real safetensors header (u64 len + JSON), no weights."""
        import struct

        header = {
            name: {"dtype": "F16", "shape": list(shape),
                   "data_offsets": [0, 0]}
            for name, shape in shapes.items()
        }
        blob = json.dumps(header).encode("utf-8")
        path = tmp_path / "model.safetensors"
        path.write_bytes(struct.pack("<Q", len(blob)) + blob)
        return tmp_path

    def test_local_checkpoint_is_measured_not_guessed(self, tmp_path):
        from soup_cli.utils.gpu import model_size_from_name

        # 2 tensors x 1e8 params = 0.2B
        self._fake_checkpoint(
            tmp_path, {"a.weight": [10000, 10000], "b.weight": [10000, 10000]}
        )
        assert model_size_from_name(str(tmp_path)) == pytest.approx(0.2)

    def test_nameless_local_dir_does_not_become_7b(self, tmp_path):
        from soup_cli.utils.gpu import model_size_from_name

        merged = tmp_path / "denseA"  # no size marker anywhere in the name
        merged.mkdir()
        self._fake_checkpoint(merged, {"w": [1000, 1000]})
        assert model_size_from_name(str(merged)) < 1.0, (
            "a nameless local checkpoint must not hit the 7B default"
        )

    def test_sharded_checkpoint_sums_all_shards(self, tmp_path):
        import struct

        from soup_cli.utils.gpu import model_size_from_name

        for idx in range(2):
            header = {
                "w": {"dtype": "F16", "shape": [10000, 5000],
                      "data_offsets": [0, 0]}
            }
            blob = json.dumps(header).encode("utf-8")
            (tmp_path / f"model-0000{idx}.safetensors").write_bytes(
                struct.pack("<Q", len(blob)) + blob
            )
        assert model_size_from_name(str(tmp_path)) == pytest.approx(0.1)

    def test_hub_ids_still_use_the_name_guess(self):
        from soup_cli.utils.gpu import model_size_from_name

        assert model_size_from_name("meta-llama/Meta-Llama-3-8B") == 8
        assert model_size_from_name(
            "HuggingFaceTB/SmolLM2-135M-Instruct"
        ) == pytest.approx(0.135)

    def test_missing_dir_falls_back_to_the_guess(self):
        from soup_cli.utils.gpu import model_size_from_name

        assert model_size_from_name("./definitely-not-here") == 7.0

    def test_dir_without_safetensors_falls_back(self, tmp_path):
        from soup_cli.utils.gpu import model_size_from_name

        (tmp_path / "config.json").write_text("{}", encoding="utf-8")
        assert model_size_from_name(str(tmp_path)) == 7.0

    def test_corrupt_header_falls_back_rather_than_crashing(self, tmp_path):
        from soup_cli.utils.gpu import model_size_from_name

        (tmp_path / "model.safetensors").write_bytes(b"\xff" * 4)  # truncated
        assert model_size_from_name(str(tmp_path)) == 7.0

    def test_absurd_header_length_is_refused(self, tmp_path):
        import struct

        from soup_cli.utils.gpu import model_size_from_name

        (tmp_path / "model.safetensors").write_bytes(
            struct.pack("<Q", 2**40) + b"{}"
        )
        assert model_size_from_name(str(tmp_path)) == 7.0


class TestNoTopLevelTorch:
    """The light core must stay importable without the [train] extra.

    v0.71.0 split the deps deliberately; a stray top-level `import torch`
    in a data module would silently undo that for every `soup data ...`
    user.
    """

    _BANNED = {"torch", "transformers", "peft"}

    @pytest.mark.parametrize(
        "module", ["semdedup", "topics", "canary", "rehearsal", "embed"]
    )
    def test_no_top_level_heavy_import(self, module):
        import ast
        import pathlib

        import soup_cli

        path = pathlib.Path(soup_cli.__file__).parent / "utils" / f"{module}.py"
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in tree.body:  # TOP LEVEL only — lazy imports are the point
            if isinstance(node, ast.Import):
                for alias in node.names:
                    assert alias.name.split(".")[0] not in self._BANNED, (
                        f"{module}.py imports {alias.name} at top level"
                    )
            elif isinstance(node, ast.ImportFrom) and node.module:
                assert node.module.split(".")[0] not in self._BANNED, (
                    f"{module}.py imports from {node.module} at top level"
                )

    def test_the_guard_can_actually_fail(self):
        """Guard the guard: prove the AST walk detects a top-level import."""
        import ast

        tree = ast.parse("import torch\n\ndef f():\n    import peft\n")
        top = [
            alias.name
            for node in tree.body
            if isinstance(node, ast.Import)
            for alias in node.names
        ]
        assert "torch" in top, "walk must see a real top-level import"
        assert "peft" not in top, "and must ignore a lazy one inside a function"

    @pytest.mark.parametrize(
        "module",
        [
            "soup_cli.utils.semdedup",
            "soup_cli.utils.topics",
            "soup_cli.utils.canary",
            "soup_cli.utils.rehearsal",
            "soup_cli.utils.embed",
            "soup_cli.commands.data_topics",
            "soup_cli.commands.data_canary",
        ],
    )
    def test_module_imports_cleanly(self, module):
        import importlib

        assert importlib.import_module(module) is not None

    def test_new_commands_are_registered(self):
        """The CLI must actually expose what this release built."""
        from typer.testing import CliRunner

        from soup_cli.cli import app

        res = CliRunner(env={"COLUMNS": "200"}).invoke(app, ["data", "--help"])
        assert res.exit_code == 0, (res.output, repr(res.exception))
        cleaned = _clean(res.output)
        assert "topics" in cleaned
        assert "canary" in cleaned