tests/test_sparse_convergence.py

"""
Tests for SparseConvergenceLoop and SparseMultiHop.

Verifies that sparse co-occurrence convergence produces the same
logical behavior as the dense convergence loop:
- Converges on related words
- Does not converge on empty/unknown input
- Query anchor prevents drift
- Per-hop specialization (early=broad, late=narrow)
- Mutual attention boosts coherent clusters
- Softmax-weighted blending
- Multi-hop finds distant concepts
- Trace is inspectable
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from sparse_convergence import (
    SparseConvergenceLoop, SparseMultiHop,
    sparse_cosine, sparse_blend, sparse_norm, sparse_normalize,
    SparseConvergenceResult, SparseMultiHopResult,
)


def make_graph():
    """Build a small co-occurrence graph for testing.

    Words: paris(0), capital(1), france(2), london(3), england(4),
           river(5), seine(6), thames(7), bridge(8), tower(9)

    Co-occurrence edges encode knowledge:
      paris <-> capital, france, seine, bridge
      london <-> capital, england, thames, tower, bridge
      capital <-> paris, london, france, england
      seine <-> paris, river
      thames <-> london, river
    """
    words = ["paris", "capital", "france", "london", "england",
             "river", "seine", "thames", "bridge", "tower"]
    word_idx = {w: i for i, w in enumerate(words)}
    word_neurons = {w: i + 100 for i, w in enumerate(words)}

    cooc = {}
    for i in range(len(words)):
        cooc[i] = {i: 1.0}

    def link(a, b, weight=0.3):
        cooc[a][b] = cooc[a].get(b, 0) + weight
        cooc[b][a] = cooc[b].get(a, 0) + weight

    # paris cluster
    link(0, 1, 0.5)  # paris-capital
    link(0, 2, 0.6)  # paris-france
    link(0, 6, 0.4)  # paris-seine
    link(0, 8, 0.2)  # paris-bridge

    # london cluster
    link(3, 1, 0.5)  # london-capital
    link(3, 4, 0.6)  # london-england
    link(3, 7, 0.4)  # london-thames
    link(3, 9, 0.3)  # london-tower
    link(3, 8, 0.2)  # london-bridge

    # shared
    link(1, 2, 0.4)  # capital-france
    link(1, 4, 0.4)  # capital-england

    # river connections
    link(5, 6, 0.5)  # river-seine
    link(5, 7, 0.5)  # river-thames

    return cooc, word_idx, words, word_neurons


def make_loop(cooc, word_idx, words, word_neurons, **kwargs):
    defaults = dict(max_hops=10, k=5, convergence_threshold=0.99,
                    min_confidence=0.05, min_relevance=0.1, temperature=1.0)
    defaults.update(kwargs)
    return SparseConvergenceLoop(
        cooc=cooc, word_idx=word_idx, words=words,
        word_neurons=word_neurons, **defaults
    )


class TestSparseConvergenceBasic:

    def test_converges_on_related_words(self):
        """Query [paris] should converge and find france/capital."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        result = loop.converge([word_idx["paris"]])

        assert result.converged is True
        assert len(result.concepts) > 0
        assert result.confidence > 0

    def test_empty_query_does_not_converge(self):
        """Empty query should not converge."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        result = loop.converge([])

        assert result.converged is False
        assert result.concepts == []

    def test_unknown_index_does_not_crash(self):
        """Index not in cooc should handle gracefully."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        result = loop.converge([999])

        assert result.converged is False

    def test_convergence_finds_correct_cluster(self):
        """Query [paris, france] should find paris-related words, not london."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        result = loop.converge([word_idx["paris"], word_idx["france"]])

        concept_indices = {widx for widx, _ in result.concepts}
        # Should find paris-cluster words
        assert word_idx["capital"] in concept_indices or word_idx["seine"] in concept_indices


class TestSparseAnchor:

    def test_anchor_keeps_query_relevant(self):
        """With query anchor, result should stay near query, not drift."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons, max_hops=10)
        result = loop.converge([word_idx["paris"]])

        if result.converged and result.hops:
            # Last hop's current profile should still have paris-related keys
            last_profile = result.hops[-1].current
            # Paris (idx 0) should still have weight in the profile
            assert last_profile.get(0, 0) > 0 or last_profile.get(2, 0) > 0

    def test_movement_decreases(self):
        """Movement should generally decrease (convergence)."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons, max_hops=10)
        result = loop.converge([word_idx["paris"]])

        if len(result.hops) >= 3:
            first = result.hops[0].movement
            last = result.hops[-1].movement
            assert last <= first + 0.05  # tolerance


class TestSparseMutualAttention:

    def test_coherent_cluster_boosted(self):
        """Words that co-occur with each other should get boosted."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)

        # paris(0) and france(2) mutually co-occur
        neighbors = [(0, 0.5), (2, 0.4), (5, 0.3)]  # paris, france, river
        boosted = loop._mutual_attention(neighbors)

        # paris and france should be boosted more than river
        paris_sim = next(s for w, s in boosted if w == 0)
        river_sim = next(s for w, s in boosted if w == 5)
        assert paris_sim > river_sim


class TestSparseSoftmaxBlend:

    def test_high_similarity_dominates(self):
        """Higher similarity word should dominate the blend."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons, temperature=0.5)

        neighbors = [(0, 0.9), (3, 0.1)]  # paris strong, london weak
        blended = loop._softmax_blend(neighbors)

        # Blended profile should lean toward paris's co-occurrences
        # paris has france(2), london has england(4)
        paris_weight = blended.get(2, 0)  # france
        london_weight = blended.get(4, 0)  # england
        assert paris_weight > london_weight

    def test_uniform_at_inf_temperature(self):
        """Infinite temperature should give uniform weighting."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons,
                         temperature=float('inf'))

        neighbors = [(0, 0.9), (3, 0.1)]
        blended = loop._softmax_blend(neighbors)
        # Both should contribute roughly equally
        assert len(blended) > 0


class TestSparseTrace:

    def test_trace_has_hops(self):
        """Trace should record each hop."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        result = loop.converge([word_idx["paris"]])

        assert len(result.hops) > 0
        for hop in result.hops:
            assert hop.hop_number >= 0
            assert len(hop.neighbors) > 0

    def test_trace_string_readable(self):
        """Trace should produce human-readable string."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        result = loop.converge([word_idx["paris"]])

        trace_str = result.trace()
        assert "SparseConvergence:" in trace_str
        assert "Hop 0:" in trace_str


class TestSparseMultiHop:

    def test_single_hop_works(self):
        """Single query should work through multi-hop."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        mh = SparseMultiHop(loop, max_rounds=3)
        result = mh.reason([word_idx["paris"]])

        assert len(result.rounds) >= 1
        assert len(result.concepts) > 0

    def test_multi_hop_discovers_distant_concept(self):
        """Multi-hop from [seine] should eventually find [london] via river->thames."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons,
                         min_relevance=0.05, min_confidence=0.01)
        mh = SparseMultiHop(loop, max_rounds=3, concept_blend_weight=0.5)
        result = mh.reason([word_idx["seine"]])

        concept_indices = {widx for widx, _ in result.concepts}
        # Should discover river or thames through the graph
        found_river_cluster = (word_idx["river"] in concept_indices or
                               word_idx["thames"] in concept_indices)
        assert found_river_cluster, f"Expected to find river/thames, got {concept_indices}"

    def test_no_duplicate_concepts(self):
        """Same concept should not appear twice across rounds."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        mh = SparseMultiHop(loop, max_rounds=3)
        result = mh.reason([word_idx["paris"]])

        indices = [widx for widx, _ in result.concepts]
        assert len(indices) == len(set(indices))

    def test_max_rounds_respected(self):
        """Should not exceed max_rounds."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        mh = SparseMultiHop(loop, max_rounds=2)
        result = mh.reason([word_idx["paris"]])

        assert len(result.rounds) <= 2

    def test_multi_hop_trace_readable(self):
        """Multi-hop trace should be human-readable."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        mh = SparseMultiHop(loop, max_rounds=3)
        result = mh.reason([word_idx["paris"]])

        trace_str = result.trace()
        assert "SparseMultiHop:" in trace_str
        assert "Round 1" in trace_str

    def test_empty_query_multi_hop(self):
        """Empty query should not crash."""
        cooc, word_idx, words, word_neurons = make_graph()
        loop = make_loop(cooc, word_idx, words, word_neurons)
        mh = SparseMultiHop(loop, max_rounds=3)
        result = mh.reason([])

        assert result.converged is False
        assert result.concepts == []


class TestSparseUtils:

    def test_sparse_cosine_identical(self):
        a = {0: 1.0, 1: 2.0}
        assert abs(sparse_cosine(a, a) - 1.0) < 1e-6

    def test_sparse_cosine_orthogonal(self):
        a = {0: 1.0}
        b = {1: 1.0}
        assert sparse_cosine(a, b) == 0.0

    def test_sparse_cosine_empty(self):
        assert sparse_cosine({}, {0: 1.0}) == 0.0
        assert sparse_cosine({}, {}) == 0.0

    def test_sparse_blend_uniform(self):
        p1 = {0: 1.0, 1: 2.0}
        p2 = {1: 4.0, 2: 6.0}
        blended = sparse_blend([p1, p2])
        assert abs(blended[0] - 0.5) < 1e-6
        assert abs(blended[1] - 3.0) < 1e-6
        assert abs(blended[2] - 3.0) < 1e-6

    def test_sparse_normalize(self):
        d = {0: 3.0, 1: 4.0}
        n = sparse_normalize(d)
        assert abs(sparse_norm(n) - 1.0) < 1e-6


if __name__ == "__main__":
    import pytest
    pytest.main([__file__, "-v"])