SkyDiscover/evox

"""EvoX Memory component — the search-strategy history H.

One file per component (see scaffold.py). Faithful port of SkyDiscover's
``SearchStrategyDatabase`` behind the :class:`~galapagos.components.memory.Memory` interface:
H stores one entry per finalized strategy — ``{source, metrics, start_stats, end_stats}`` where
``metrics`` is the :class:`~galapagos.scaffolds.evox.scaffold.LogWindowScorer` output (J as
``combined_score`` + window metadata) and the stats are the raw φ snapshots recorded at
deployment/finalization (``start_db_stats``/``end_db_stats``).

Meta-selection mirrors ``SearchStrategyDatabase.sample``: :meth:`best` is the deterministic
argmax-J parent (``max`` with a ``-inf`` guard for non-numeric scores — first entry wins ties);
:meth:`inspirations` samples up to ``n`` entries uniformly at random then filters out the best
(so it may return fewer than ``n``, exactly like upstream). Unbounded on purpose — runs produce
on the order of ten strategies (one per stagnation event).

Memory interface mapping: ``read()`` renders a short human summary of H;
``write(_, kind="strategy", entry=...)`` records one finalized strategy (the scaffold also calls
:meth:`add_strategy` directly).
"""
from __future__ import annotations

import random

from ...components.memory import Memory


def _safe_score(entry: dict) -> float:
    score = entry.get("metrics", {}).get("combined_score")
    return float(score) if isinstance(score, (int, float)) else float("-inf")


class EvoXStrategyMemory(Memory):
    """The strategy history H — argmax-J parent selection + uniform-random inspirations."""

    def __init__(self) -> None:
        self._entries: list[dict] = []

    # ---- Memory interface ----------------------------------------------------------------
    def read(self, spec: dict | None = None) -> str:
        if not self._entries:
            return ""
        lines = []
        for i, entry in enumerate(self._entries, 1):
            m = entry.get("metrics", {})
            lines.append(
                f"- strategy {i}: J={m.get('combined_score', 0.0):.4f}, "
                f"window start iter {m.get('window_start_iteration') or 0}, "
                f"horizon {m.get('search_horizon') or 0}, "
                f"score {m.get('search_window_start_score', 0.0):.4f} -> "
                f"{m.get('search_window_end_score', 0.0):.4f}")
        return "\n".join(lines)

    def write(self, knowledge: str, **meta) -> None:
        if meta.get("kind") == "strategy" and isinstance(meta.get("entry"), dict):
            self.add_strategy(meta["entry"])

    # ---- SearchStrategyDatabase port --------------------------------------------------------
    def add_strategy(self, entry: dict) -> None:
        """Record one finalized strategy: {source, metrics, start_stats, end_stats}."""
        self._entries.append(entry)

    def best(self) -> dict | None:
        """The argmax-J strategy (the greedy meta-parent); first entry wins ties."""
        if not self._entries:
            return None
        return max(self._entries, key=_safe_score)

    def inspirations(self, n: int = 2, rng: random.Random | None = None) -> list[dict]:
        """Up to ``n`` uniform-random strategies, excluding the best (sampled-then-filtered like
        upstream, so the result may hold fewer than ``n``)."""
        rng = rng or random.Random(0)
        pool = list(self._entries)
        num = max(0, min(int(n), len(pool)))
        picked = rng.sample(pool, num) if num > 0 else []
        best = self.best()
        return [entry for entry in picked if entry is not best]

    @property
    def entries(self) -> list[dict]:
        return list(self._entries)

    def __len__(self) -> int:
        return len(self._entries)

    def __bool__(self) -> bool:
        """Always truthy: an EMPTY history is still a real Memory (the base scaffold falls back
        to ``NullMemory`` on a falsy ``memory`` argument, and ``__len__`` alone would make a
        fresh history falsy)."""
        return True