tests/test_steering.py

# tests/test_steering.py
"""Plan 6, tâche 3 — steering organique : les cartes dirigent la bouche.

``Brain.speak_bias(question, tokenizer)`` : ask_cards → PREMIER token BPE de
chaque réponse de carte (convention GPT-2 : espace initial, ``" paris"``) →
``{token_id: boost décroissant par rang}`` — le dictionnaire qui entre dans
``speak(..., bias_tokens=...)`` (les organes orientent la bouche).

Le cerveau est SUBSTRAT (numpy seul, gate mingw) : le tokenizer est passé EN
PARAMÈTRE — aucun import torch/tokenizers ici au niveau module. Les tests
substrat tournent donc dans les DEUX venvs ; les tests BPE (lib `tokenizers`)
et l'intégration ``speak()`` (torch) se skipent proprement en mingw.
"""
import pytest

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"),
]
BOOKS = [Atom("who wrote hamlet", "william shakespeare", "books.csv:2")]

# Ordre des cartes pour la question france (vérifié, D=2048) :
# [HEBBIEN paris, FAIT paris, FAIT rome, FAIT madrid] — rangs 0..3.


def _brain() -> Brain:
    brain = Brain(D=2048)
    brain.ingest_source("capitals", CAPITALS)
    brain.ingest_source("books", BOOKS)
    return brain


class FakeAnswerTok:
    """Tokenizer fake du protocole minimal : le PREMIER id encode le mot
    entier préfixé d'un espace (« paris » → [7, 8] comme un vrai BPE qui
    fusionnerait). Injectif sur les ids connus → les poids par token sont
    observables sans ambiguïté."""

    vocab_size = 128
    eos_token_id = 127
    _WORDS = {
        " paris": [7, 8],
        " madrid": [9],
        " tokyo": [10],
        " rome": [11],
    }

    def encode(self, text: str) -> list[int]:
        return list(self._WORDS.get(text, [1]))

    def decode(self, ids: list[int]) -> str:
        rev = {v[0]: k.strip() for k, v in self._WORDS.items()}
        return "".join(rev.get(int(i), "?") for i in ids)


# ---------------------------------------------------------------------------
# speak_bias — substrat (les deux venvs)
# ---------------------------------------------------------------------------

def test_speak_bias_contains_paris_token_positive_weight():
    """La question france → le premier token de " paris" est dans le bias,
    avec le boost MAXIMAL (rang 0 : HEBBIEN/FAIT paris)."""
    bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok())
    assert bias[7] == pytest.approx(8.0)  # boost par défaut, rang 0
    assert all(isinstance(t, int) for t in bias)
    assert all(isinstance(w, float) for w in bias.values())
    assert all(0 <= t < FakeAnswerTok.vocab_size for t in bias)


def test_speak_bias_max_not_sum():
    """HEBBIEN (rang 0) et FAIT top-1 (rang 1) donnent la MÊME réponse
    "paris" → accumulation MAX : le poids reste boost (8.0), PAS 8+4=12 —
    l'échelle du steering reste bornée par boost quel que soit l'accord
    des cartes (calibrée contre l'échelle z-norm, std 3)."""
    bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok())
    assert bias[7] == pytest.approx(8.0)
    assert bias[7] < 12.0  # pas de sommation


def test_speak_bias_weights_decay_by_rank():
    """Rangs 2 et 3 (FAIT rome, madrid) : boost × 0.5² et × 0.5³ — les poids
    décroissent strictement avec le rang, tous < au poids du rang 0."""
    bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok())
    assert bias[11] == pytest.approx(8.0 * 0.25)  # rome, rang 2
    assert bias[9] == pytest.approx(8.0 * 0.125)  # madrid, rang 3
    others = [w for t, w in bias.items() if t != 7]
    assert others and all(w < bias[7] for w in others)


def test_speak_bias_boost_scales_ladder():
    """Le ladder entier est proportionnel à boost : {8, 2, 1} à boost 8,
    {50, 12.5, 6.25} à boost 50 (rang 0/2/3, rang 1 fusionné par max)."""
    bias = _brain().speak_bias("what is the capital of france", FakeAnswerTok())
    assert sorted(bias.values(), reverse=True) == pytest.approx([8.0, 2.0, 1.0])
    big = _brain().speak_bias(
        "what is the capital of france", FakeAnswerTok(), boost=50.0
    )
    assert big[7] == pytest.approx(50.0)
    assert big[11] == pytest.approx(12.5)


def test_speak_bias_lacune_gives_empty_bias():
    """Cerveau vide → seule carte LACUNE (texte placeholder, pas une
    réponse) → bias vide : on ne steere jamais la bouche avec une lacune."""
    empty = Brain(D=2048)
    assert empty.speak_bias("anything", FakeAnswerTok()) == {}


def test_speak_bias_deterministic():
    """Même question, même cerveau → même dict (ordre d'insertion inclus) :
    le steering est reproductible, comme toute la parole mécanique."""
    brain = _brain()
    a = brain.speak_bias("what is the capital of france", FakeAnswerTok())
    b = brain.speak_bias("what is the capital of france", FakeAnswerTok())
    assert a == b and list(a) == list(b)


def test_speak_bias_rejects_bad_arguments():
    brain = _brain()
    with pytest.raises(ValueError):
        brain.speak_bias("q", FakeAnswerTok(), boost=0.0)
    with pytest.raises(ValueError):
        brain.speak_bias("q", FakeAnswerTok(), boost=-1.0)
    with pytest.raises(ValueError):
        brain.speak_bias("q", FakeAnswerTok(), k=0)


# ---------------------------------------------------------------------------
# BPE réel (lib tokenizers — se skipe en substrat mingw)
# ---------------------------------------------------------------------------

def test_speak_bias_with_real_gpt2_bpe():
    """Le VRAI tokenizer GPT-2 : le bias contient le premier token de
    " paris" (convention GPT-2 : l'espace initial fait partie du token —
    encode(' paris')[0] != encode('paris')[0]) avec un poids > 0."""
    bpt = pytest.importorskip("fractus_vorax.model.bpe_tokenizer")
    if not bpt.bpe_available():
        pytest.skip("tokenizers indisponible")
    tok = bpt.Gpt2BpeTokenizer()
    # la convention espace-initial est réelle : deux ids différents
    assert tok.encode(" paris")[0] != tok.encode("paris")[0]

    bias = _brain().speak_bias("what is the capital of france", tok)
    paris_id = tok.encode(" paris")[0]
    assert paris_id in bias
    assert bias[paris_id] > 0.0
    assert bias[paris_id] == pytest.approx(8.0)  # rang 0, max
    assert all(0 <= t < tok.vocab_size for t in bias)


# ---------------------------------------------------------------------------
# Intégration : speak(tiny, ..., bias_tokens=speak_bias(...)) (torch)
# ---------------------------------------------------------------------------

def test_speak_runs_with_organic_bias_deterministic():
    """Intégration mini du plan : speak() avec le bias organique s'exécute
    et est déterministe (même seed ⇒ même texte, diagnostics inclus)."""
    torch = pytest.importorskip(
        "torch", reason="speak nécessite torch (substrat sans torch)"
    )
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig
    from fractus_vorax.model.speak import speak

    torch.manual_seed(0)
    core = CteCore(CteCoreConfig(vocab_size=128))
    tok = FakeAnswerTok()
    bias = _brain().speak_bias("what is the capital of france", tok)
    assert bias  # non vide, ids dans le vocab tiny (<= 128)
    assert all(0 <= t < 128 for t in bias)

    text_a, diags_a = speak(
        core, tok, "what is the capital of france",
        max_new_tokens=4, bias_tokens=bias, seed=0,
    )
    text_b, diags_b = speak(
        core, tok, "what is the capital of france",
        max_new_tokens=4, bias_tokens=bias, seed=0,
    )
    assert text_a == text_b and diags_a == diags_b
    assert isinstance(text_a, str)


def test_speak_strong_organic_bias_emits_answer_first():
    """Boost fort (50, le δ prouvé dominateur en tâche 1) : le PREMIER token
    émis EST le token-réponse des organes (le id 7 = " paris") — les organes
    dirigent la bouche, mesurable pas seulement exécutable."""
    torch = pytest.importorskip(
        "torch", reason="speak nécessite torch (substrat sans torch)"
    )
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig
    from fractus_vorax.model.speak import speak

    torch.manual_seed(0)
    core = CteCore(CteCoreConfig(vocab_size=128))
    tok = FakeAnswerTok()
    bias = _brain().speak_bias(
        "what is the capital of france", tok, boost=50.0
    )
    for seed in range(10):
        _, diags = speak(
            core, tok, "hello", max_new_tokens=1,
            bias_tokens=bias, seed=seed,
        )
        assert diags[0]["chosen"] == tok.decode([7]), (
            f"seed {seed}: les organes n'ont pas dirigé le premier pas"
        )