tests/test_speak_integration.py

# tests/test_speak_integration.py
"""Plan 6, tâche 4 — l'intégration : ``:speak`` dans le REPL, core_speak v2.

Substrat (mingw, TOUJOURS vert) : ``:speak`` sans noyau attaché → la ligne
honnête ``[PAROLE] noyau non attaché (:core d'abord)`` — jamais de traceback,
le substrat ne touche pas torch.

Torch (se skipe en mingw) : un tiny CteCore + tokenizer char attachés au
REPL → les trois lignes ``[PAROLE]`` (cartes, steering, parole mécanique) ;
``run_core_speak(mode=...)`` sur un tiny checkpoint fake → le dict v2 avec
greedy/mécanique (et steerée avec brain) + le ciel ouvert (routage, têtes).
"""
import pytest

from fractus_vorax.agent.repl import Repl
from fractus_vorax.brain import Brain
from fractus_vorax.compiler.atoms import Atom

CAPITALS = [
    Atom("what is the capital of france", "paris", "cap.csv:2"),
    Atom("what is the capital of spain", "madrid", "cap.csv:3"),
    Atom("what is the capital of japan", "tokyo", "cap.csv:4"),
    Atom("what is the capital of italy", "rome", "cap.csv:5"),
]

DEGRADE = "[PAROLE] noyau non attaché (:core d'abord)"


def _repl(tmp_path) -> Repl:
    brain = Brain(D=2048)
    brain.ingest_source("capitals", CAPITALS)
    brain.save(tmp_path / "brain")
    return Repl(tmp_path / "brain", D=2048)


class _Tok128:
    """Tokenizer char-level sur le vocab tiny 128 (protocole minimal)."""

    vocab_size = 128

    def encode(self, text: str) -> list[int]:
        return [ord(c) % 128 for c in text]

    def decode(self, ids: list[int]) -> str:
        return "".join(chr(int(i) % 128) for i in ids)


# ---------------------------------------------------------------------------
# :speak — dégradation honnête (substrat, les deux venvs)
# ---------------------------------------------------------------------------
def test_speak_without_core_degrades_honestly(tmp_path):
    """Sans :core : UNE ligne honnête, jamais de traceback — le contrat
    mingw du plan (skip-proof : ce test ne touche jamais torch)."""
    r = _repl(tmp_path)
    assert r.core is None and r.tokenizer is None
    out = r.feed(":speak hello")
    assert out == [DEGRADE]
    # le REPL reste vivant après la dégradation (pas d'état cassé)
    assert any("[CARTE]" in l for l in r.feed("what is the capital of japan"))


def test_speak_usage_line_without_text(tmp_path):
    r = _repl(tmp_path)
    assert r.feed(":speak") == ["[PAROLE] usage: :speak <texte>"]


def test_speak_with_fake_tiny_core(tmp_path):
    """Noeud torch : tiny CteCore + tokenizer char attachés → les TROIS
    lignes ``[PAROLE]`` (cartes, steering, parole) — la boucle mécanique
    vit dans le REPL, pilotée par les organes."""
    torch = pytest.importorskip(
        "torch", reason=":speak mécanique nécessite torch (substrat sans torch)"
    )
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig

    r = _repl(tmp_path)
    torch.manual_seed(0)
    r.core = CteCore(CteCoreConfig(vocab_size=128))
    r.tokenizer = _Tok128()
    out = r.feed(":speak what is the capital of france")
    assert any(l.startswith("[PAROLE] cartes: ") for l in out)
    steering = [l for l in out if l.startswith("[PAROLE] steering:")]
    assert steering, f"ligne steering absente: {out}"
    # les organes steered : le premier token char de ' paris' (l'espace, 32)
    assert " x8" in steering[0]
    speeches = [l for l in out if l.startswith('[PAROLE] "')]
    assert speeches, f"ligne parole absente: {out}"
    # déterminisme : même seed (7, interne au :speak) => même parole
    again = r.feed(":speak what is the capital of france")
    assert [l for l in again if l.startswith('[PAROLE] "')] == speeches


def test_speak_tiny_core_failure_is_honest(tmp_path):
    """Un noyau qui échoue (tokenizer incohérent avec le vocab) → ligne
    ``[PAROLE] indisponible: ...``, jamais de traceback."""
    torch = pytest.importorskip(
        "torch", reason=":speak mécanique nécessite torch (substrat sans torch)"
    )
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig

    r = _repl(tmp_path)
    torch.manual_seed(0)
    r.core = CteCore(CteCoreConfig(vocab_size=8))  # vocab minuscule
    r.tokenizer = _Tok128()                        # ids jusqu'à 127 -> hors vocab
    out = r.feed(":speak what is the capital of france")
    assert any(l.startswith("[PAROLE] indisponible:") for l in out)


# ---------------------------------------------------------------------------
# core_speak v2 — run function sur un tiny checkpoint fake (torch)
# ---------------------------------------------------------------------------
def _tiny_ckpt(tmp_path):
    torch = pytest.importorskip("torch")
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig

    # vocab BPE complet (50257) : le harnais tokenize en vrai GPT-2 — les
    # ids doivent vivre dans le vocab du noyau pour que forward passe.
    torch.manual_seed(0)
    core = CteCore(CteCoreConfig(vocab_size=50257))
    ckpt = tmp_path / "tiny_cte.pt"
    torch.save({"model_state": core.state_dict()}, ckpt)
    return ckpt


def _bpe_or_skip():
    from fractus_vorax.model.bpe_tokenizer import bpe_available

    if not bpe_available():
        pytest.skip("tokenizers indisponible")
    try:
        from fractus_vorax.model.bpe_tokenizer import Gpt2BpeTokenizer

        Gpt2BpeTokenizer()
    except Exception as exc:  # cache HF vide + réseau injoignable
        pytest.skip(f"tokenizer GPT-2 indisponible: {exc}")


def test_core_speak_mechanic_mode_tiny(tmp_path):
    """mode mechanic : baseline greedy + parole mécanique par question, ciel
    ouvert (routage top-2/couche + têtes) sur la première question."""
    _bpe_or_skip()
    from bench.core_speak import run_core_speak

    result = run_core_speak(
        _tiny_ckpt(tmp_path),
        ["what is the capital of france", "what is the capital of spain"],
        brain=None,
        max_new_tokens=3,
        mode="mechanic",
    )
    assert result["mode"] == "mechanic"
    assert len(result["runs"]) == 2
    for run in result["runs"]:
        assert isinstance(run["greedy_output"], str)
        assert isinstance(run["mechanic_output"], str)
        assert isinstance(run["mechanic_diags"], list) and run["mechanic_diags"]
        assert run["mechanic_answer_token"] is False  # sans brain -> pas d'attendu
    # ciel ouvert : top-2 experts par couche (2 couches sur le tiny), têtes lues
    assert set(result["routing"]) == {"layer_0", "layer_1"}
    for entries in result["routing"].values():
        assert len(entries) == 2 and abs(sum(w for _, w in entries) - 1.0) < 1e-5
    conf_sal = result["head_readout"]
    assert 0.0 <= conf_sal["confidence"] < 1.0 and 0.0 <= conf_sal["salience"] < 1.0
    assert result["answer_token_rate_steered"] is None  # pas de steered en mechanic


def test_core_speak_steered_mode_tiny(tmp_path):
    """mode steered (exige brain) : les TROIS générations par question —
    greedy attracteur, mécanique non-steerée, mécanique steerée (même seed,
    bias organique) — et les deux taux de tokens-réponse mesurés."""
    _bpe_or_skip()
    from bench.core_speak import run_core_speak

    brain = Brain(D=2048)
    brain.ingest_source("capitals", CAPITALS)
    brain.save(tmp_path / "brain")
    result = run_core_speak(
        _tiny_ckpt(tmp_path),
        ["what is the capital of france", "what is the capital of spain"],
        brain=tmp_path / "brain",
        max_new_tokens=3,
        mode="steered",
    )
    assert result["mode"] == "steered"
    for run in result["runs"]:
        for key in ("greedy_output", "mechanic_output", "steered_output"):
            assert isinstance(run[key], str)
        assert run["bias_tokens"], "steering organique vide alors que le brain sait"
        assert isinstance(run["steered_answer_token"], bool)
        assert run["expected_answer"] in ("paris", "madrid")
        assert isinstance(run["answer_token"], int)
    assert isinstance(result["answer_token_rate_mechanic"], float)
    assert isinstance(result["answer_token_rate_steered"], float)
    assert result["verdict"] in ("word salad", "answers present")


def test_core_speak_steered_without_brain_is_refused(tmp_path):
    """steered sans brain : précondition refusée proprement (ValueError du
    run, exit 1 du CLI) — pas d'organes, pas de steering."""
    _bpe_or_skip()
    from bench.core_speak import main, run_core_speak

    ckpt = _tiny_ckpt(tmp_path)
    with pytest.raises(ValueError):
        run_core_speak(ckpt, ["q"], brain=None, max_new_tokens=2, mode="steered")
    code = main(["--ckpt", str(ckpt), "--mode", "steered"])
    assert code == 1