"""PALIMPSESTE — Append-only knowledge base ``M`` (Axiomes 1, 3, 4). Axiome 1 — Tout est adresse A knowledge item is a triplet ``(a, v, w)`` where: - ``a`` is the address (a bound, context-sensitive key hypervector) - ``v`` is the content (a bound value hypervector) - ``w`` is a confidence weight in ``R+`` (decays with time, never zero) Axiome 3 — Apprendre est une ecriture ``M <- M ∪ {(a=bind(x,c), v=y, w=1)}`` costs ``O(1)`` amortized: a single append to a table. This module guarantees that invariant. Axiome 4 — Les parametres sont reconstruits, non stockes ``M`` never stores a weight matrix; it stores *traces* that the read-back kernel ``Phi`` (see ``phi.py``) reconstructs parameters from on demand. Design ------ ``M`` is an append-only log of :class:`Trace` records, plus an :class:`LSHIndex` over the addresses for sub-linear neighborhood retrieval. Crucially: - nothing is ever *deleted* — forgetting is a *soft* decay of the access weight ``w`` (a dormant memory can re-awaken when its address is queried); - insert is ``O(1)`` amortized (one list append + L LSH bucket appends); - the LSH index may occasionally be rebuilt to re-tune (K, L) as ``|M|`` grows; rebuilding never drops traces. A reserved *meta-subspace* ``H_meta`` is a tag on certain traces whose address lies in a reserved region of address-space. The ``MetaController`` (``meta.py``) reads/writes there to rewrite its own read-back kernel. """ from __future__ import annotations from dataclasses import dataclass, field import math import time import numpy as np from .hv import HV from .lsh import LSHConfig, LSHIndex __all__ = ["Trace", "Memory", "MemoryStats"] @dataclass(frozen=True) class Trace: """An immutable knowledge record in ``M``. Attributes ---------- id : int Position in the append log (also the LSH item id). address : HV Bound key ``a = bind(x, c)`` used for associative lookup. value : HV Bound content ``v`` that ``Phi`` recovers for matching addresses. weight : float Confidence / access weight ``w`` in ``(0, 1]``. Decays softly over time; never reaches 0 (a floor is enforced) so dormant memories can re-awaken. t_insert : float Wall-clock insertion time (monotonic), used for soft decay. meta : bool If True, this trace lives in the reserved meta-subspace ``H_meta`` and is interpreted by the ``MetaController`` rather than ordinary recall. tag : str | None Optional human-readable label for inspection/debugging. """ id: int address: HV value: HV weight: float = 1.0 t_insert: float = field(default_factory=time.monotonic) meta: bool = False tag: str | None = None def __post_init__(self) -> None: if self.id < 0: raise ValueError("trace id must be >= 0") if self.weight <= 0.0: raise ValueError("weight must be > 0 (append-only: no zeroing)") @dataclass class MemoryStats: """Lightweight stats snapshot of ``M``.""" n_traces: int n_meta: int mean_weight: float min_weight: float lsh_size: int @dataclass class Memory: """The append-only knowledge base ``M``. Parameters ---------- D : int Hypervector dimensionality (must match addresses/values). lsh_config : LSHConfig | None Index tuning. If None, auto-tuned for ~10% radius at 0.9 recall. decay : dict Soft-forgetting parameters for ``current_weight``: - ``half_life`` : wall-clock seconds for ``w`` to halve (default inf). - ``floor`` : minimum weight floor (default 1e-3). rng : np.random.Generator For reproducible LSH projections. """ D: int lsh_config: LSHConfig | None = None decay: dict = field(default_factory=lambda: {"half_life": math.inf, "floor": 1e-3}) rng: np.random.Generator = field(default_factory=np.random.default_rng) _traces: list[Trace] = field(default_factory=list) _meta_traces: list[Trace] = field(default_factory=list) _index: LSHIndex | None = None _meta_index: LSHIndex | None = None def __post_init__(self) -> None: if self.lsh_config is None: self.lsh_config = LSHConfig.tune(D=self.D, target_radius=0.10, recall=0.9) elif self.lsh_config.D != self.D: raise ValueError("lsh_config.D must match Memory.D") self._index = LSHIndex(config=self.lsh_config, _rng=self.rng) # --------------------------------------------------------------- capacity def __len__(self) -> int: return len(self._traces) @property def traces(self) -> list[Trace]: """All (non-meta) traces in insertion order.""" return self._traces @property def meta_traces(self) -> list[Trace]: """Traces in the reserved ``H_meta`` subspace.""" return self._meta_traces @property def index(self) -> LSHIndex: assert self._index is not None return self._index # ----------------------------------------------------------------- insert def write( self, address: HV, value: HV, weight: float = 1.0, meta: bool = False, tag: str | None = None, ) -> Trace: """Append a trace ``(a, v, w)`` to ``M``. O(1) amortized. This is the *only* mutation primitive. Nothing is ever deleted. """ if address.D != self.D or value.D != self.D: raise ValueError( f"address/value D must equal Memory.D={self.D}" ) if meta: tid = len(self._meta_traces) + 10_000_000 # disjoint id space tr = Trace( id=tid, address=address, value=value, weight=weight, meta=True, tag=tag, ) self._meta_traces.append(tr) # meta traces are indexed in a *separate* index to keep the main # recall space clean of self-rewriting noise. self._ensure_meta_index().insert(tr.id, address) return tr tid = len(self._traces) tr = Trace( id=tid, address=address, value=value, weight=weight, meta=False, tag=tag, ) self._traces.append(tr) assert self._index is not None self._index.insert(tid, address) return tr # -------------------------------------------------------------- retrieval def candidates(self, query: HV) -> list[int]: """Return LSH candidate trace ids for ``query`` (pre-Hamming-filter).""" assert self._index is not None return sorted(self._index.query_candidates(query)) def current_weight(self, tr: Trace, now: float | None = None) -> float: """Soft-decayed weight of a trace at time ``now``. ``w_now = floor + (w0 - floor) * 2^(-(t-t0)/half_life)``. With ``half_life = inf`` (default) this is constant ``w0``. """ if now is None: now = time.monotonic() hl = self.decay.get("half_life", math.inf) floor = self.decay.get("floor", 1e-3) if math.isinf(hl): return tr.weight elapsed = max(0.0, now - tr.t_insert) decayed = tr.weight * (0.5 ** (elapsed / hl)) return max(floor, decayed) def stats(self) -> MemoryStats: ws = [t.weight for t in self._traces] return MemoryStats( n_traces=len(self._traces), n_meta=len(self._meta_traces), mean_weight=float(np.mean(ws)) if ws else 0.0, min_weight=float(np.min(ws)) if ws else 0.0, lsh_size=self._index.size if self._index else 0, ) # ------------------------------------------------------------- meta index def _ensure_meta_index(self) -> LSHIndex: if self._meta_index is None: self._meta_index = LSHIndex(config=self.lsh_config, _rng=self.rng) return self._meta_index def meta_candidates(self, query: HV) -> list[int]: """Candidate meta-trace ids for ``query`` in ``H_meta``.""" if self._meta_index is None: return [] # Map the meta ids back to meta_traces positions for the caller. return sorted(self._meta_index.query_candidates(query)) def get_meta(self, meta_id: int) -> Trace | None: for tr in self._meta_traces: if tr.id == meta_id: return tr return None # ------------------------------------------------------------------ io def rebuild_index(self) -> None: """Rebuild the LSH index (e.g. after changing K/L). Never drops traces.""" assert self._index is not None self._index.rebuild([t.address for t in self._traces]) if self._meta_index is not None: self._meta_index.rebuild([t.address for t in self._meta_traces])