# 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 "] # --------------------------------------------------------------------------- # 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)