tests/test_answer_lock.py

# tests/test_answer_lock.py
"""Plan 7 — le verrou de réponse : les organes articulent à travers le noyau.

Quatre angles :
- ``speak(..., answer_lock, answer_boost)`` : à l'étape i, +boost sur
  answer_lock[i] (APRÈS z-norm/rep-pen, AVANT top-k/p) — boost par défaut
  30, CALIBRÉ PAR MESURE (le 12 du plan perd le tirage quand le token
  verrouillé vit à −1σ et le top à +4σ : z[' par']=−3.0 vs zmax=11.9 sur
  le 1B — 2/4 capitales seulement ; le tiny core non-entraîné porte les
  mêmes outliers froids, zmax 17-20) : le token verrouillé est choisi à
  CHAQUE étape i, 10 seeds sur 10 ;
- ``Brain.speak_answer`` : la liste COMPLÈTE des tokens BPE de " "+réponse
  (tokyo pour japan), ANALOGIE en repli, LACUNE -> [] ;
- ``:say`` sans noyau : la dégradation honnête (mingw-safe, jamais torch) ;
- intégration tiny : REPL + tiny CteCore -> les lignes [ORGANES]/[PAROLE],
  la réponse verrouillée VISIBLE dans la génération.
"""
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"),
]


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


class FakeAnswerTok:
    """Tokenizer fake du protocole minimal (même table que test_steering) :
    « paris » -> [7, 8] comme un vrai BPE qui fusionnerait — la liste
    COMPLÈTE est observable, pas juste le premier geste."""

    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)


class _TinyCharTok:
    """Tokenizer char-level sur le vocab tiny (128) — injectif sur les ids < 128."""

    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(answer_lock=...) — le verrou domine chaque pas (torch, skip en mingw)
# ---------------------------------------------------------------------------
def _torch_tiny():
    torch = pytest.importorskip(
        "torch", reason="speak nécessite torch (substrat sans torch)"
    )
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig

    torch.manual_seed(0)
    return torch, CteCore(CteCoreConfig(vocab_size=128))


def test_answer_lock_locks_every_step_ten_seeds():
    """Boost par défaut (30, calibré par mesure sur le 1B — le 12 du plan
    perdait contre les outliers zmax≈12-20 alors que z[locked] peut vivre à
    −3σ) : à CHAQUE étape i le token émis EST answer_lock[i], 10 seeds sur
    10 — la réponse des organes passe À TRAVERS la distribution du noyau,
    pas à côté."""
    _, core = _torch_tiny()
    from fractus_vorax.model.speak import speak

    tok = _TinyCharTok()
    lock = [42, 84, 7]  # ids distincts : le rep-pen ne touche jamais le verrou
    for seed in range(10):
        _, diags = speak(
            core, tok, "hello", max_new_tokens=3,
            answer_lock=lock, seed=seed,
        )
        assert [d["chosen"] for d in diags] == [tok.decode([i]) for i in lock], (
            f"seed {seed}: le verrou n'a pas tenu à chaque pas"
        )


def test_answer_lock_boost_twelve_is_honestly_weak():
    """Le 12 du plan, gardé comme MESURE honnête : sur le tiny core non
    entraîné (outliers zmax≈18 froids), +12 sur un token à −1σ NE domine
    PAS — c'est la mesure qui a motivé le défaut 30 (voir speak.py). Ce
    test fige la frontière : le boost place le token, il ne le force pas
    magiquement."""
    _, core = _torch_tiny()
    from fractus_vorax.model.speak import speak

    tok = _TinyCharTok()
    text12, _ = speak(
        core, tok, "hello", max_new_tokens=1,
        answer_lock=[42], answer_boost=12.0, seed=0,
    )
    text30, _ = speak(
        core, tok, "hello", max_new_tokens=1,
        answer_lock=[42], answer_boost=30.0, seed=0,
    )
    assert text30 == tok.decode([42])  # 30 : le verrou tient (même seed)
    assert text12 != tok.decode([42])  # 12 : z[42]−1.2+12 < zmax 18.4 — perdu


def test_answer_lock_survives_top_k_mask():
    """Le verrou vit AVANT top-k/p : le token boosté est top-1, il survit au
    masque — top_k=5 ne casse pas l'articulation."""
    _, core = _torch_tiny()
    from fractus_vorax.model.speak import speak

    tok = _TinyCharTok()
    for seed in range(5):
        _, diags = speak(
            core, tok, "hello", max_new_tokens=2,
            answer_lock=[42, 84], top_k=5, seed=seed,
        )
        assert [d["chosen"] for d in diags] == [tok.decode([42]), tok.decode([84])]


def test_answer_lock_longer_than_generation_is_fine():
    """len(answer_lock) > max_new_tokens : le verrou s'arrête avec la boucle,
    aucune erreur, les pas émis sont verrouillés."""
    _, core = _torch_tiny()
    from fractus_vorax.model.speak import speak

    tok = _TinyCharTok()
    text, diags = speak(
        core, tok, "hello", max_new_tokens=2,
        answer_lock=[42, 84, 7, 90, 55], seed=0,
    )
    assert len(diags) == 2
    assert [d["chosen"] for d in diags] == [tok.decode([42]), tok.decode([84])]
    assert isinstance(text, str)


def test_answer_lock_rejects_bad_arguments():
    _, core = _torch_tiny()
    from fractus_vorax.model.speak import speak

    tok = _TinyCharTok()
    with pytest.raises(ValueError):  # id hors vocab
        speak(core, tok, "hello", max_new_tokens=2, answer_lock=[999])
    with pytest.raises(ValueError):  # boost nul ou négatif
        speak(core, tok, "hello", max_new_tokens=2, answer_lock=[42], answer_boost=0.0)


def test_answer_lock_visible_in_top3_diagnostics():
    """Le boost est posé AVANT les diagnostics top3 : le token verrouillé
    y apparaît (observabilité du mécanisme — il est affiché, pas caché)."""
    _, core = _torch_tiny()
    from fractus_vorax.model.speak import speak

    tok = _TinyCharTok()
    _, diags = speak(core, tok, "hello", max_new_tokens=1, answer_lock=[42], seed=0)
    top3_tokens = [s for s, _v in diags[0]["top3"]]
    assert tok.decode([42]) in top3_tokens
    assert top3_tokens[0] == tok.decode([42])  # top-1 après le boost


# ---------------------------------------------------------------------------
# Brain.speak_answer — substrat (les deux venvs)
# ---------------------------------------------------------------------------
def test_speak_answer_japan_is_full_tokyo_list():
    """japan -> la liste COMPLÈTE de " tokyo" (pas juste le premier token —
    c'est la différence avec speak_bias : UN verrou, TOUS ses tokens)."""
    assert _brain().speak_answer("what is the capital of japan", FakeAnswerTok()) == [10]


def test_speak_answer_france_is_full_paris_list():
    """france -> [7, 8] : les DEUX tokens de " paris" — le verrou couvre la
    réponse entière, fragment après fragment."""
    assert _brain().speak_answer("what is the capital of france", FakeAnswerTok()) == [7, 8]


def test_speak_answer_lacune_is_empty():
    """Cerveau vide -> seule carte LACUNE -> [] : on ne verrouille JAMAIS la
    bouche sur une lacune."""
    assert Brain(D=2048).speak_answer("anything", FakeAnswerTok()) == []


def test_speak_answer_rejects_bad_k():
    with pytest.raises(ValueError):
        _brain().speak_answer("q", FakeAnswerTok(), k=0)


def test_speak_answer_deterministic():
    brain = _brain()
    a = brain.speak_answer("what is the capital of japan", FakeAnswerTok())
    b = brain.speak_answer("what is the capital of japan", FakeAnswerTok())
    assert a == b == [10]


def test_speak_answer_with_real_gpt2_bpe():
    """Le VRAI BPE GPT-2 : japan -> exactement encode(" tokyo"), liste
    complète (se skipe en substrat mingw sans lib tokenizers)."""
    bpt = pytest.importorskip("fractus_vorax.model.bpe_tokenizer")
    if not bpt.bpe_available():
        pytest.skip("tokenizers indisponible")
    tok = bpt.Gpt2BpeTokenizer()
    assert _brain().speak_answer("what is the capital of japan", tok) == tok.encode(" tokyo")
    assert _brain().speak_answer("what is the capital of france", tok) == tok.encode(" paris")


# ---------------------------------------------------------------------------
# :say — dégradation honnête (substrat, les deux venvs)
# ---------------------------------------------------------------------------
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)


def test_say_without_core_degrades_honestly(tmp_path):
    """Sans :core : UNE ligne honnête, jamais de traceback — le contrat
    mingw (ce test ne touche jamais torch)."""
    r = _repl(tmp_path)
    assert r.core is None and r.tokenizer is None
    assert r.feed(":say what is the capital of japan") == [
        "[PAROLE] noyau non attaché (:core d'abord)"
    ]
    # le REPL reste vivant après la dégradation
    assert any("[CARTE]" in l for l in r.feed("what is the capital of japan"))


def test_say_usage_line_without_question(tmp_path):
    r = _repl(tmp_path)
    assert r.feed(":say") == ["[PAROLE] usage: :say <question>"]


# ---------------------------------------------------------------------------
# Intégration tiny — :say complet sur un tiny CteCore (torch)
# ---------------------------------------------------------------------------
def test_say_tiny_core_organes_and_locked_parole(tmp_path):
    """Noeud torch : tiny CteCore + tokenizer char -> les DEUX lignes.
    [ORGANES] porte la carte (paris), [PAROLE] porte le prompt
    d'articulation ET la génération où la réponse verrouillée est VISIBLE :
    les 6 tokens char de " paris" passent À TRAVERS la boucle du noyau."""
    torch = pytest.importorskip(
        "torch", reason=":say 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 = _TinyCharTok()
    out = r.feed(":say what is the capital of france")
    organes = [l for l in out if l.startswith("[ORGANES]")]
    parole = [l for l in out if l.startswith("[PAROLE]")]
    assert len(organes) == 1 and len(parole) == 1, f"lignes manquantes: {out}"
    assert "paris" in organes[0]  # la carte qui connaît : stmt => paris
    assert parole[0].startswith(
        "[PAROLE] what is the capital of france The answer is"
    )
    # LE point du Plan 7 : la réponse des organes apparaît dans la parole
    # du noyau (verrou = " paris" char par char, puis suite libre)
    assert " paris" in parole[0]
    # déterminisme : même seed interne (7) -> même articulation
    again = r.feed(":say what is the capital of france")
    assert [l for l in again if l.startswith("[PAROLE]")] == parole


def test_say_tiny_core_lacune_says_no_lock(tmp_path):
    """Cerveau SANS connaissance (aucun atome -> carte LACUNE) : la parole
    coule SANS verrou et le DIT — honnête, jamais de verrou sur rien.
    (Un brain non vide répond toujours des FAIT bas-sim : la lacune vraie
    est le cerveau vide.)"""
    torch = pytest.importorskip(
        "torch", reason=":say mécanique nécessite torch (substrat sans torch)"
    )
    from fractus_vorax.model.cte_core import CteCore, CteCoreConfig

    r = Repl(tmp_path / "brain-vide", D=2048)  # pas de meta.json -> brain vide
    torch.manual_seed(0)
    r.core = CteCore(CteCoreConfig(vocab_size=128))
    r.tokenizer = _TinyCharTok()
    out = r.feed(":say what is the capital of france")
    parole = [l for l in out if l.startswith("[PAROLE]")]
    assert parole and "sans verrou" in parole[0]
    assert any("LACUNE" in l for l in out)  # la lacune est dite comme lacune


def test_say_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=":say 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 = _TinyCharTok()                   # ids jusqu'à 127 -> hors vocab
    out = r.feed(":say what is the capital of france")
    assert any(l.startswith("[PAROLE] indisponible:") for l in out)