# tests/test_brain_v2.py import numpy as np from fractus_vorax.brain import ANALOGY_TRIGGER_SIM, Brain from fractus_vorax.cards import Card from fractus_vorax.compiler.atoms import Atom from fractus_vorax.compiler.encode import text_hv from fractus_vorax.hv import bind 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")] def _brain(): brain = Brain(D=2048) brain.ingest_source("capitals", CAPITALS) brain.ingest_source("books", BOOKS) return brain def test_ingest_source_spawns_expert_and_relations(): brain = _brain() assert len(brain.experts) == 2 assert brain.experts.get("capitals") is not None assert any("capital" in " ".join(p) for p in brain.relations) assert brain.hebbian.n_writes >= 5 def test_ask_cards_fact(): brain = _brain() cards = brain.ask_cards("what is the capital of japan", k=2) facts = [c for c in cards if c.kind == "FAIT"] assert facts and "tokyo" in facts[0].text assert facts[0].sim > 0.5 def test_ask_cards_analogy_on_typo(): brain = _brain() cards = brain.ask_cards("what is the capital of franc", k=3) ana = [c for c in cards if c.kind == "ANALOGIE"] assert ana and "paris" in ana[0].text def test_consolidate_creates_composite(): brain = _brain() for _ in range(3): brain.ask_cards("what is the capital of france", k=2) n0 = len(brain) promoted = brain.consolidate(threshold=3) assert promoted >= 1 assert len(brain) == n0 + promoted comp = brain.atoms[-1] assert comp.provenance == "consolidated" def test_consolidate_is_idempotent(): brain = _brain() for _ in range(3): brain.ask_cards("what is the capital of france", k=2) n1 = brain.consolidate(threshold=3) n2 = brain.consolidate(threshold=3) assert n1 >= 1 and n2 == 0 # plus de re-promotion def test_plan1_compat_ask_still_works(): brain = _brain() hits = brain.ask("what is the capital of spain", k=1) assert hits[0][0].context == "madrid" def test_save_load_roundtrip_v2(tmp_path): brain = _brain() brain.save(tmp_path / "brain") loaded = Brain.load(tmp_path / "brain") assert len(loaded.experts) == 2 assert loaded.hebbian.n_writes == brain.hebbian.n_writes assert loaded.ask("who wrote hamlet", k=1)[0][0].context == "william shakespeare" def test_load_plan1_brain_still_works(tmp_path): """Un brain Plan 1 (sans organes persistés) doit se charger.""" brain = Brain(D=2048) brain.ingest_atoms(CAPITALS) brain.save(tmp_path / "old") # simuler l'absence des fichiers v2 import shutil for sub in ("experts", "hebbian"): shutil.rmtree(tmp_path / "old" / sub, ignore_errors=True) (tmp_path / "old" / "coactivation.json").unlink(missing_ok=True) (tmp_path / "old" / "relations.json").unlink(missing_ok=True) loaded = Brain.load(tmp_path / "old") assert len(loaded) == 4 assert len(loaded.experts) == 0 def test_analogy_trigger_is_named_constant(): """Le seuil d'analogie est nommé et calibré : typos ~0.63-0.69, exact à 1.0.""" assert 0.6 < ANALOGY_TRIGGER_SIM < 0.9 def test_consolidate_roundtrip_preserves_composite_addr(tmp_path): """save/load doit rejouer addr = bind(addr_a, addr_b), pas text_hv(statement).""" brain = _brain() for _ in range(3): brain.ask_cards("what is the capital of france", k=2) assert brain.consolidate(threshold=3) >= 1 comp_id = len(brain) - 1 a_id, b_id = brain._composite_parents[comp_id] addr_before = brain.traces.addrs[comp_id].copy() expected = bind(brain.traces.addrs[a_id], brain.traces.addrs[b_id]) assert np.array_equal(addr_before, expected) query = text_hv("what is the capital of france", brain.D, brain.seed) top_before = brain.traces.retrieve(query, k=3)[0][0] brain.save(tmp_path / "brain") loaded = Brain.load(tmp_path / "brain") assert np.array_equal(loaded.traces.addrs[-1], addr_before) assert loaded._composite_parents[comp_id] == (a_id, b_id) assert np.array_equal( loaded.traces.addrs[comp_id], bind(loaded.traces.addrs[a_id], loaded.traces.addrs[b_id]), ) # le retrieval retrace pareil : même top-1 qu'avant la sauvegarde assert loaded.traces.retrieve(query, k=3)[0][0] == top_before def _force_depth2_composite(brain: Brain, comp_id: int, threshold: int = 3) -> None: """Fait entrer le composite (profondeur 1) dans une paire co-activée. Les addr bind étant quasi orthogonales à tout, une requête hors-domaine les classe dans le bruit proche de zéro : le top-k déterministe peut inclure le composite. On essaie quelques (requête, k) ; sinon fallback white-box : brain.coact.record() direct — même compteur qu'alimente ask_cards(), acceptable et documenté. """ for query, k in [("zzz quantum lorem ipsum", 2), ("zzz quantum lorem ipsum", 3), ("xyzw quokka blorb", 2), ("xyzw quokka blorb", 3)]: for _ in range(threshold): brain.ask_cards(query, k=k) if any(c >= threshold and comp_id in pair for pair, c in brain.coact.counts.items()): return for _ in range(threshold): brain.coact.record([comp_id, 0]) def test_nested_composite_roundtrip_preserves_addr(tmp_path): """Composites imbriqués (profondeur ≥ 2) : addr bit-identique au save/load. consolidate() lie les addr STOCKÉES des parents (le bind d'un parent composite n'est pas re-productible par encode_atom), donc addr = bind(addr_a_stockée, addr_b_stockée) à toute profondeur — exactement le replay de Brain.load(). Avant le fix, le ré-encodage divergeait dès la profondeur 2. Étape « force top-k » : voir _force_depth2_composite — white-box fallback documenté si la requête hors-domaine ne classe pas le composite. """ brain = _brain() # profondeur 1 : co-activation répétée sur une question du domaine for _ in range(3): brain.ask_cards("what is the capital of france", k=2) assert brain.consolidate(threshold=3) >= 1 comp_id = len(brain) - 1 assert comp_id in brain._composite_parents # profondeur 2 : le composite doit co-activer avec un autre atome _force_depth2_composite(brain, comp_id) n_before = len(brain) brain.consolidate(threshold=3) assert len(brain) > n_before, "la consolidation de profondeur 2 n'a pas eu lieu" d2 = [aid for aid, (pa, pb) in brain._composite_parents.items() if aid > comp_id and comp_id in (pa, pb)] assert d2, "aucun composite de profondeur 2 créé (parents ne contiennent pas comp_id)" d2_id = d2[0] p_a, p_b = brain._composite_parents[d2_id] addr_before = brain.traces.addrs[d2_id].copy() brain.save(tmp_path / "brain") loaded = Brain.load(tmp_path / "brain") assert np.array_equal(loaded.traces.addrs[d2_id], addr_before) assert np.array_equal( loaded.traces.addrs[d2_id], bind(loaded.traces.addrs[p_a], loaded.traces.addrs[p_b]), )