examples/quickstart.py

"""PALIMPSESTE — End-to-end integration demo.

Runs the full autonomous loop on a small deterministic environment (a ring of
states) and prints telemetry showing:
  - surprise decreasing as the agent learns the transition
  - memory growing append-only (no retraining)
  - consolidation producing abstract concepts
  - the meta-controller occasionally rewriting its kernel under the Lyapunov
    constraint (audit-logged in H_meta)

Run:  python examples/quickstart.py
"""

from __future__ import annotations

import numpy as np

from palimseste import hv
from palimseste.loop import Palimseste, Environment, LoopConfig


class RingEnv(Environment):
    """A ring of N states; each tick advances to the next state.

    Adjacent states share most bits, so the transition o_t -> o_{t+1} is
    learnable by associative memory.
    """

    def __init__(self, D: int, n_states: int = 5, seed: int = 0):
        self.D = D
        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 // 80)
        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)

    def observe(self) -> hv.HV:
        return self.states[self._i]

    def actions(self) -> list[hv.HV]:
        return [self._advance, hv.random_hv(D=self.D)]  # 2 actions: advance / noop

    def act(self, action: hv.HV) -> None:
        # action 0 (advance) moves the ring; action 1 (noop) stays
        if action is self._advance:
            self._i = (self._i + 1) % len(self.states)

    def done(self) -> bool:
        return False


def main() -> None:
    D = 3000
    agent = Palimseste(
        D=D,
        rng=np.random.default_rng(0),
        loop_cfg=LoopConfig(
            surprise_threshold=0.2,
            consolidate_every=24,
            meta_every=96,
            max_radius=150,
        ),
    )
    env = RingEnv(D=D, n_states=5, seed=0)

    print("=" * 72)
    print("PALIMPSESTE — autonomous active-inference loop demo")
    print(f"D={D}  ring_states=5  ticks=600")
    print("=" * 72)

    surprises: list[float] = []
    for t in range(1, 601):
        r = agent.step(env)
        surprises.append(r.surprise)
        if t % 100 == 0 or t == 1:
            recent = np.mean(surprises[max(0, t - 50):t])
            stats = agent.stats()
            print(
                f"t={t:4d}  surprise={r.surprise:.3f}  "
                f"mean(50)={recent:.3f}  |M|={stats['n_traces']:5d}  "
                f"concepts={stats['n_concepts']:3d}  "
                f"meta_dec={stats['n_meta_decisions']:3d}  "
                f"kernel={stats['kernel']}"
            )
            if r.consolidation is not None and r.consolidation.promoted:
                print(f"         consolidated {len(r.consolidation.promoted)} concept(s)")
            if r.meta is not None and r.meta.accepted:
                print(
                    f"         META rewrite ACCEPTED: "
                    f"ΔL={r.meta.delta:+.4f} -> {r.meta.proposal.config.encode()}"
                )

    print("=" * 72)
    first = np.mean(surprises[:50])
    last = np.mean(surprises[-50:])
    print(f"surprise: first-50 mean = {first:.3f}   last-50 mean = {last:.3f}")
    print(f"Δ = {last - first:+.3f}  ({'decreased ✓' if last < first else 'NOT decreased ✗'})")
    stats = agent.stats()
    print(f"final |M| = {stats['n_traces']} traces, "
          f"{stats['n_concepts']} concepts, "
          f"{stats['n_meta']} meta-traces (H_meta audit log)")
    print("=" * 72)


if __name__ == "__main__":
    main()