tests/test_brain_v2.py

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