File size: 7,192 Bytes
9cae168
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Tests for the meta-controller & Lyapunov energy (``palimseste.meta``).

Core guarantees under test:
  - a proposal that reduces surprise is accepted; one that increases it is rejected
  - accepted rewrites are audit-logged into H_meta (append-only)
  - invariants penalize configs (e.g. max_radius blocks runaway radius)
  - the acceptance criterion itself is never rewritten (constitution immutability)
  - build_replay draws from M's own traces
"""

from __future__ import annotations

import numpy as np
import pytest

from palimseste import hv
from palimseste.memory import Memory
from palimseste.phi import Phi, KernelConfig
from palimseste.meta import (
    MetaController,
    LyapunovEnergy,
    MetaProposal,
    Invariant,
    max_radius_invariant,
    _config_to_hv,
)


def _setup(D=2000, seed=1, radius=5, n_traces=40):
    rng = np.random.default_rng(seed)
    mem = Memory(D=D, rng=np.random.default_rng(seed))
    # populate M with (address, value) pairs where value == address+noise,
    # so a wider radius genuinely helps reconstruction (lower surprise).
    base = hv.random_hv(D=D, rng=rng)
    for _ in range(n_traces):
        a = _flip(base, 50, rng)
        v = _flip(a, 3, rng)  # value near its address
        mem.write(a, v)
    phi = Phi(config=KernelConfig(radius=radius, min_weight=1e-6))
    energy = LyapunovEnergy()
    ctrl = MetaController(mem=mem, phi=phi, energy=energy, rng=rng)
    return mem, phi, ctrl, rng


def _flip(h: hv.HV, n: int, rng) -> hv.HV:
    s = hv.bits_to_signs(h)
    pos = rng.choice(h.D, size=n, replace=False)
    s[pos] = -s[pos]
    return hv.signs_to_bits(s)


def _noisy_replay(mem, n, flip_bits, rng):
    """Replay set of (noisy_query, target_value): query is a stored address
    with ``flip_bits`` positions flipped, so radius=0 misses but a wider
    radius finds the true value. This is the *generalization* regime where
    the kernel's radius actually matters."""
    idx = rng.choice(len(mem.traces), size=min(n, len(mem.traces)), replace=False)
    replay = []
    for i in idx:
        tr = mem.traces[i]
        q = _flip(tr.address, flip_bits, rng)
        replay.append((q, tr.value))
    return replay


def test_evaluate_accepts_surprise_reducing_proposal():
    mem, phi, ctrl, rng = _setup(radius=0, n_traces=40)
    # noisy queries: 15 bits flipped -> radius 0 misses, radius 20 hits
    replay = _noisy_replay(mem, 40, flip_bits=15, rng=rng)
    prop = MetaProposal(KernelConfig(radius=20, min_weight=1e-6), "widen radius")
    dec = ctrl.evaluate(prop, replay)
    assert dec.accepted
    assert dec.delta < 0


def test_evaluate_rejects_surprise_increasing_proposal():
    mem, phi, ctrl, rng = _setup(radius=30, n_traces=40)
    # with radius 30, noisy queries are found; shrinking to 0 breaks them
    replay = _noisy_replay(mem, 40, flip_bits=15, rng=rng)
    prop = MetaProposal(KernelConfig(radius=0, min_weight=1e-6), "shrink radius")
    dec = ctrl.evaluate(prop, replay)
    assert not dec.accepted
    assert dec.delta > 0


def test_accepted_rewrite_is_audit_logged_to_hmeta():
    mem, phi, ctrl, rng = _setup(radius=0, n_traces=30)
    n_meta_before = len(mem.meta_traces)
    replay = ctrl.build_replay(30)
    prop = MetaProposal(KernelConfig(radius=15, min_weight=1e-6), "widen")
    dec = ctrl.evaluate(prop, replay)
    assert dec.accepted
    # commit manually (evaluate doesn't commit; step does)
    phi.config = dec.proposal.config
    ctrl._write_current_config()
    assert len(mem.meta_traces) == n_meta_before + 1


def test_step_commits_first_accepted_proposal():
    mem, phi, ctrl, rng = _setup(radius=0, n_traces=30)
    replay = ctrl.build_replay(30)
    cfg_before = phi.config
    dec = ctrl.step(replay, max_proposals=20)
    # at least one widening proposal should be accepted eventually
    assert dec is not None
    if dec.accepted:
        assert phi.config == dec.proposal.config
        assert phi.config != cfg_before


def test_invariant_max_radius_blocks_runaway():
    # with a tiny max_radius invariant, widening beyond it is penalized
    mem, phi, ctrl, rng = _setup(radius=0, n_traces=20)
    ctrl.energy = LyapunovEnergy(invariants=[max_radius_invariant(max_r=5)])
    replay = ctrl.build_replay(20)
    # propose radius 100 -> invariant violation huge -> rejected
    prop = MetaProposal(KernelConfig(radius=100, min_weight=1e-6), "runaway")
    dec = ctrl.evaluate(prop, replay)
    assert not dec.accepted
    assert "rejected" in dec.reason


def test_invariant_zero_when_satisfied():
    inv = max_radius_invariant(max_r=50)
    mem = Memory(D=500, rng=np.random.default_rng(0))
    cfg_ok = KernelConfig(radius=10)
    cfg_bad = KernelConfig(radius=100)
    assert inv.violation(mem, cfg_ok) == 0.0
    assert inv.violation(mem, cfg_bad) > 0.0


def test_constitution_immutability():
    # The acceptance criterion (LyapunovEnergy + invariants) is NOT stored in
    # H_meta and cannot be rewritten by the controller. Verify the controller
    # has no API to mutate its own energy/invariants.
    mem, phi, ctrl, rng = _setup(radius=5, n_traces=10)
    # The controller exposes `energy` as a field, but there is no method to
    # *write* a new energy into H_meta. The only thing it writes to H_meta is
    # kernel configs. Assert the meta-traces are all tagged kernel_config:*.
    for tr in mem.meta_traces:
        assert tr.tag is not None and tr.tag.startswith("kernel_config:")


def test_build_replay_from_empty_memory():
    mem = Memory(D=500, rng=np.random.default_rng(0))
    phi = Phi(config=KernelConfig(radius=5))
    ctrl = MetaController(mem=mem, phi=phi, energy=LyapunovEnergy(),
                          rng=np.random.default_rng(0))
    assert ctrl.build_replay(10) == []


def test_config_to_hv_deterministic_and_distinct():
    D = 1000
    rng = np.random.default_rng(0)
    a = KernelConfig(radius=5, min_weight=1e-3, sharpness=0.0, topk=None)
    b = KernelConfig(radius=5, min_weight=1e-3, sharpness=0.0, topk=None)
    c = KernelConfig(radius=6, min_weight=1e-3, sharpness=0.0, topk=None)
    ha = _config_to_hv(a, D, rng)
    hb = _config_to_hv(b, D, rng)
    hc = _config_to_hv(c, D, rng)
    assert ha == hb
    assert ha != hc


def test_history_records_all_evaluated_proposals():
    mem, phi, ctrl, rng = _setup(radius=0, n_traces=15)
    replay = ctrl.build_replay(15)
    ctrl.step(replay, max_proposals=5)
    # at least 1, at most 5 decisions recorded
    assert 1 <= len(ctrl.history) <= 5


def test_custom_invariant_callable():
    called = {"n": 0}

    def _viol(_mem, _cfg):
        called["n"] += 1
        return 0.0

    inv = Invariant(name="custom", violation=_viol, lam=2.0)
    energy = LyapunovEnergy(invariants=[inv])
    mem = Memory(D=500, rng=np.random.default_rng(0))
    phi = Phi(config=KernelConfig(radius=2))
    ctrl = MetaController(mem=mem, phi=phi, energy=energy, rng=np.random.default_rng(0))
    # write a trace so replay is non-empty
    rng = np.random.default_rng(1)
    mem.write(hv.random_hv(D=500, rng=rng), hv.random_hv(D=500, rng=rng))
    replay = ctrl.build_replay(1)
    ctrl.evaluate(MetaProposal(KernelConfig(radius=3), "test"), replay)
    assert called["n"] >= 1