File size: 2,176 Bytes
dbb04e4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c3a3710
dbb04e4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c3a3710
 
 
dbb04e4
 
 
 
 
 
c3a3710
dbb04e4
 
 
 
 
 
c3a3710
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
"""

Test HAIMEngine Synapse Cleanup

"""
import os
import pytest
from mnemocore.core.engine import HAIMEngine
from mnemocore.core.synapse import SynapticConnection
from mnemocore.core.config import reset_config

@pytest.fixture
def test_engine(tmp_path):
    reset_config()
    data_dir = tmp_path / "data"
    data_dir.mkdir()

    synapses_file = data_dir / "synapses.json"
    os.environ["HAIM_DATA_DIR"] = str(data_dir)
    os.environ["HAIM_MEMORY_FILE"] = str(data_dir / "memory.jsonl")
    os.environ["HAIM_SYNAPSES_FILE"] = str(synapses_file)

    engine = HAIMEngine()
    yield engine

    # Cleanup
    if "HAIM_DATA_DIR" in os.environ:
        del os.environ["HAIM_DATA_DIR"]
    if "HAIM_MEMORY_FILE" in os.environ:
        del os.environ["HAIM_MEMORY_FILE"]
    if "HAIM_SYNAPSES_FILE" in os.environ:
        del os.environ["HAIM_SYNAPSES_FILE"]
    reset_config()

@pytest.mark.asyncio
async def test_cleanup_decay(test_engine):
    await test_engine.initialize()
    # Add dummy synapses
    # Synapse 1: Weak (below threshold 0.1)
    syn1 = SynapticConnection("mem_1", "mem_2", initial_strength=0.05)
    test_engine.synapses[("mem_1", "mem_2")] = syn1

    # Synapse 2: Strong (above threshold 0.1)
    syn2 = SynapticConnection("mem_3", "mem_4", initial_strength=0.2)
    test_engine.synapses[("mem_3", "mem_4")] = syn2

    # Check initial count
    assert len(test_engine.synapses) == 2

    # Run cleanup
    await test_engine.cleanup_decay(threshold=0.1)

    # Verify results
    assert len(test_engine._synapse_index) == 1
    assert test_engine._synapse_index.get("mem_3", "mem_4") is not None
    assert test_engine._synapse_index.get("mem_1", "mem_2") is None

    # Verify persistence
    assert os.path.exists(test_engine.synapse_path)

@pytest.mark.asyncio
async def test_cleanup_no_decay(test_engine):
    await test_engine.initialize()
    # All strong
    syn1 = SynapticConnection("mem_1", "mem_2", initial_strength=0.5)
    test_engine.synapses[("mem_1", "mem_2")] = syn1

    await test_engine.cleanup_decay(threshold=0.1)

    assert len(test_engine._synapse_index) == 1