SkyDiscover/adaevolve

"""AdaEvolve Memory component — Level-3 paradigm (solution-tactics) state.

One file per component (see scaffold.py). Faithful port of SkyDiscover's ``ParadigmTracker``
(search/adaevolve/paradigm/tracker.py) behind the :class:`~galapagos.components.memory.Memory`
interface. Per the reference code (not the paper), stagnation is the *windowed improvement rate*:
the binary window (size 10) of "did this admitted child set a new global best" is full, no active
non-exhausted paradigm exists, and the rate is below 0.12 — not "G ≤ τ_M for all islands".

Memory interface mapping:

* ``read()``                      → the active paradigm's "## BREAKTHROUGH IDEA" injection block
  (or ``""``), consumed by the PromptBuilder.
* ``read({"spec": "tried"})``     → the formatted previously-tried ideas with SUCCESS/FAILED
  outcomes, fed back to the paradigm generator.
* ``write(_, kind="improvement", improved=..., best_score=...)`` → one improvement-window tick
  (the open mirror of ``record_improvement``).
* ``write(_, kind="paradigms", paradigms=[...], best_score=...)`` → install a fresh paradigm set
  (``set_paradigms``), archiving outgoing used paradigms with their outcome.

The rotation/usage methods (:meth:`use_paradigm`, :meth:`has_active_paradigm`,
:meth:`is_stagnating`) are concrete extras called by the scaffold.
"""
from __future__ import annotations

from ...components.memory import Memory

_PARADIGM_BLOCK = """## BREAKTHROUGH IDEA - IMPLEMENT THIS

The search has stagnated globally. You MUST implement this breakthrough idea:

**IDEA:** {idea}

**HOW TO IMPLEMENT:**
{description}

**TARGET METRIC:** {what_to_optimize}

**CAUTIONS:** {cautions}

**APPROACH TYPE:** {approach_type}

**CRITICAL:**
- You MUST implement the breakthrough idea
- Ensure the paradigm is actually used in your implementation (not just mentioned in comments)
- Correctness is essential - your implementation must be correct and functional
- Verify output format matches evaluator requirements
- Make purposeful changes that implement the idea
- Test your implementation logic carefully"""


class AdaEvolveParadigmMemory(Memory):
    """``ParadigmTracker``: bounded binary improvement window, paradigm rotation with bounded
    uses, and an outcome-annotated history of tried ideas."""

    def __init__(self, window_size: int = 10, improvement_threshold: float = 0.12,
                 max_uses: int = 2, max_tried: int = 10, num_to_generate: int = 3):
        self.window_size = int(window_size)
        self.improvement_threshold = float(improvement_threshold)
        self.max_uses = int(max_uses)
        self.max_tried = int(max_tried)
        self.num_to_generate = int(num_to_generate)
        self.improvement_history: list[float] = []      # bounded binary window
        self.active_paradigms: list[dict] = []
        self.usage_counts: list[int] = []
        self.current_index = 0
        self.tried_paradigms: list[str] = []            # bounded outcome history
        self.best_score_at_generation = 0.0
        self.best_score_during = 0.0

    # ---- Memory interface --------------------------------------------------------------------
    def read(self, spec: dict | None = None) -> str:
        if spec and spec.get("spec") == "tried":
            return "\n".join(self.tried_paradigms)
        paradigm = self.current_paradigm()
        if paradigm is None:
            return ""
        return _PARADIGM_BLOCK.format(
            idea=paradigm.get("idea", ""),
            description=paradigm.get("description", ""),
            what_to_optimize=paradigm.get("what_to_optimize", ""),
            cautions=paradigm.get("cautions", ""),
            approach_type=paradigm.get("approach_type", ""),
        )

    def write(self, knowledge: str, **meta) -> None:
        kind = meta.get("kind")
        if kind == "improvement":
            self.record_improvement(bool(meta.get("improved")), float(meta.get("best_score", 0.0)))
        elif kind == "paradigms":
            self.set_paradigms(list(meta.get("paradigms") or []),
                               float(meta.get("best_score", 0.0)))

    # ---- ParadigmTracker port ------------------------------------------------------------------
    def record_improvement(self, improved: bool, current_best_score: float) -> None:
        """One tick of the binary window — called once per admitted non-migrant child."""
        self.improvement_history.append(1.0 if improved else 0.0)
        if len(self.improvement_history) > self.window_size:
            self.improvement_history.pop(0)
        if self.active_paradigms:  # track the best score reached during the paradigms' tenure
            self.best_score_during = max(self.best_score_during, current_best_score)

    def get_improvement_rate(self) -> float:
        if not self.improvement_history:
            return 0.0
        return sum(self.improvement_history) / len(self.improvement_history)

    def is_stagnating(self) -> bool:
        """Trigger condition: full window AND no active non-exhausted paradigm AND rate below the
        threshold (the reference-code divergence from the paper's G-based trigger)."""
        return (len(self.improvement_history) >= self.window_size
                and not self.has_active_paradigm()
                and self.get_improvement_rate() < self.improvement_threshold)

    def has_active_paradigm(self) -> bool:
        """True iff a non-exhausted paradigm exists (auto-rotates past exhausted ones)."""
        if not self.active_paradigms:
            return False
        if self.usage_counts[self.current_index] >= self.max_uses:
            self._rotate()
        return self.usage_counts[self.current_index] < self.max_uses

    def current_paradigm(self) -> dict | None:
        if not self.has_active_paradigm():
            return None
        return self.active_paradigms[self.current_index]

    def use_paradigm(self) -> None:
        """Increment the current paradigm's usage, then rotate round-robin to the next
        non-exhausted paradigm."""
        if not self.has_active_paradigm():
            return
        self.usage_counts[self.current_index] += 1
        self._rotate()

    def _rotate(self) -> None:
        """Round-robin past exhausted paradigms (may stay put if only the current has uses left)."""
        n = len(self.active_paradigms)
        for offset in range(1, n + 1):
            j = (self.current_index + offset) % n
            if self.usage_counts[j] < self.max_uses:
                self.current_index = j
                return

    def set_paradigms(self, paradigms: list[dict], current_best_score: float) -> None:
        """Install a fresh set; archive outgoing *used* paradigms with outcome SUCCESS iff the
        best score rose > 0.001 during their tenure (bounded history of ``max_tried``)."""
        improvement = self.best_score_during - self.best_score_at_generation
        for i, paradigm in enumerate(self.active_paradigms):
            if self.usage_counts[i] > 0:
                outcome = "SUCCESS" if improvement > 0.001 else "FAILED"
                self.tried_paradigms.append(
                    f"{outcome}: {paradigm.get('approach_type', '?')} - {paradigm.get('idea', '?')}"
                    f" (improvement: {improvement:+.4f})")
        self.tried_paradigms = self.tried_paradigms[-self.max_tried:]
        self.active_paradigms = list(paradigms)
        self.usage_counts = [0] * len(self.active_paradigms)
        self.current_index = 0
        self.best_score_at_generation = current_best_score
        self.best_score_during = current_best_score

    def get_previously_tried_ideas(self) -> list[str]:
        return list(self.tried_paradigms)