File size: 6,490 Bytes
7156b42 | 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 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | """Tests for the autonomous loop (``palimseste.loop``).
Uses a tiny deterministic environment: a 4-state ring where each state's
observation HV is near the next state's (so the loop can learn the transition).
Verifies:
- the loop runs without error for many ticks
- surprise decreases over time as the agent learns the transition
- traces accumulate in M (append-only growth)
- StepReport telemetry is populated correctly
- curiosity picks an action (non-None) when actions are available
- reset_state clears recurrent state but not M
"""
from __future__ import annotations
import numpy as np
import pytest
from palimseste import hv
from palimseste.loop import (
Palimseste,
Environment,
StateProjector,
LoopConfig,
StepReport,
)
from palimseste.learner import Encoder
# ----------------------------------------------------------------- test env
class RingEnv(Environment):
"""A ring of N states; each tick advances to the next state.
Observations are HVs that are *near* their neighbors (a few bits apart),
so the transition o_t -> o_{t+1} is learnable by the associative memory.
Actions are [advance] (deterministic) — kept minimal so the curiosity
machinery is exercised without complicating the dynamics.
"""
def __init__(self, D: int, n_states: int = 4, seed: int = 0):
self.D = D
self.n_states = n_states
rng = np.random.default_rng(seed)
base = hv.random_hv(D=D, rng=rng)
self.states: list[hv.HV] = [base]
signs = hv.bits_to_signs(base)
bps = max(1, D // 50)
cur = signs.copy()
for _ in range(n_states - 1):
flip = rng.choice(D, size=bps, replace=False)
cur = cur.copy()
cur[flip] = -cur[flip]
self.states.append(hv.signs_to_bits(cur))
self._i = 0
self._advance = hv.random_hv(D=D, rng=rng)
self._t = 0
self._max_t = 10_000
def observe(self) -> hv.HV:
return self.states[self._i]
def actions(self) -> list[hv.HV]:
return [self._advance]
def act(self, action: hv.HV) -> None:
# single action: advance the ring
self._i = (self._i + 1) % self.n_states
self._t += 1
def done(self) -> bool:
return self._t >= self._max_t
# ----------------------------------------------------------------- tests
def _agent(D=1500, seed=0, **kw) -> Palimseste:
return Palimseste(
D=D,
rng=np.random.default_rng(seed),
loop_cfg=LoopConfig(
surprise_threshold=0.25,
consolidate_every=16,
meta_every=64,
max_radius=80,
),
**kw,
)
def test_loop_runs_many_ticks():
agent = _agent(D=1200, seed=1)
env = RingEnv(D=1200, n_states=4, seed=1)
reports = []
for _ in range(200):
reports.append(agent.step(env))
assert len(reports) == 200
assert all(isinstance(r, StepReport) for r in reports)
# memory grew (append-only learning)
assert agent.stats()["n_traces"] > 0
def test_surprise_decreases_over_time():
# After enough ticks the agent should predict the ring transition well,
# so mean surprise in the second half < mean surprise in the first half.
agent = _agent(D=1500, seed=2)
env = RingEnv(D=1500, n_states=4, seed=2)
surprises = []
for _ in range(400):
r = agent.step(env)
surprises.append(r.surprise)
first = np.mean(surprises[:100])
second = np.mean(surprises[300:])
assert second < first, f"surprise did not decrease: {first=} {second=}"
def test_step_report_fields_populated():
agent = _agent(D=1000, seed=3)
env = RingEnv(D=1000, n_states=3, seed=3)
r = agent.step(env)
assert r.t == 1
assert 0.0 <= r.surprise <= 1.0
assert r.action_idx is not None # at least one action
assert r.n_traces >= 0
assert r.n_concepts == 0 # nothing consolidated yet on tick 1
def test_curiosity_picks_action():
agent = _agent(D=1000, seed=4)
env = RingEnv(D=1000, n_states=3, seed=4)
r = agent.step(env)
assert r.action_idx == 0 # only one action available
def test_consolidation_fires():
# With a small consolidate_every, consolidation should run at least once
agent = _agent(D=1200, seed=5)
agent.loop_cfg.consolidate_every = 8
env = RingEnv(D=1200, n_states=4, seed=5)
ran_cons = False
for _ in range(40):
r = agent.step(env)
if r.consolidation is not None:
ran_cons = True
assert ran_cons
def test_meta_fires():
agent = _agent(D=1200, seed=6)
agent.loop_cfg.meta_every = 16
env = RingEnv(D=1200, n_states=4, seed=6)
ran_meta = False
for _ in range(80):
r = agent.step(env)
if r.meta is not None:
ran_meta = True
assert ran_meta
def test_reset_state_clears_recurrence_not_memory():
agent = _agent(D=1000, seed=7)
env = RingEnv(D=1000, n_states=3, seed=7)
for _ in range(30):
agent.step(env)
n_before = agent.stats()["n_traces"]
agent.reset_state()
n_after = agent.stats()["n_traces"]
assert n_before == n_after # M untouched
assert agent.surprise == 0.0 # reset clears last surprise
def test_loop_config_invalid():
with pytest.raises(ValueError):
LoopConfig(surprise_threshold=1.5)
with pytest.raises(ValueError):
LoopConfig(consolidate_every=0)
with pytest.raises(ValueError):
LoopConfig(max_radius=0)
def test_stats_keys():
agent = _agent(D=800, seed=8)
env = RingEnv(D=800, n_states=3, seed=8)
for _ in range(10):
agent.step(env)
s = agent.stats()
for k in ("t", "n_traces", "n_meta", "n_concepts", "n_meta_decisions",
"last_surprise", "kernel"):
assert k in s
def test_state_projector_reset():
enc = Encoder(D=500, rng=np.random.default_rng(0))
sp = StateProjector(D=500, encoder=enc, window=3)
o = hv.random_hv(D=500)
a = hv.random_hv(D=500)
# project (state from history, empty at first), then commit o
s_empty = sp.project(o, a)
sp.commit(o)
s_one = sp.project(o, a) # now history has o
sp.reset()
s_after = sp.project(o, a) # history cleared again
# s_empty (no history) and s_one (one item in history) differ
assert hv.similarity(s_empty, s_one) < 1.0
# after reset, state matches the empty-history state
assert hv.similarity(s_empty, s_after) == pytest.approx(1.0)
|