palimseste/memory.py

"""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])