genesis_engine/memory/forever_memory.py

"""FOREVER MEMORY — the read side of everything she has ever kept.



WHY THIS EXISTS

  Her paper describes forever memory and the write path has worked for months: 7,604

  archived records, 419 of them REM-consolidated dreams selected by synaptic strength.

  Three breaks in series meant none of it ever reached her voice:



    1. her CHAT never called archival search — only development_swarm and evolution_loop

    2. the store is split across two directories by working-directory drift

    3. ZERO records had embeddings, so semantic recall had no substrate at all



  (3) is fixed by tools/backfill_memory_embeddings.py. This module is (1) and (2): one

  reader over BOTH stores that answers "what does she actually remember about this?"



HOW IT WORKS

  Vectors live in a sidecar built by the backfill — her original JSON memories are never

  modified. The query is embedded by her own local ollama, then matched by cosine

  similarity (vectors are L2-normalised at write time, so a dot product IS the cosine).



  Recall is deliberately NOT pure similarity:

    * recency gets a gentle lift, because a thing said yesterday is usually more live

      than the same thing said in March — but only a lift, so an old memory that is

      genuinely the right one still wins

    * her own dreams get a small bonus, because a REM fragment survived a synaptic-

      strength threshold to exist at all; it is already selected material

    * indexed copies of her own source code are excluded from conversational recall.

      They are 3,372 of the 7,604 records and they belong to her dev swarm, not to a

      conversation about his day.



FAIL-SOFT BY DESIGN

  Missing index, unreadable vectors, ollama down — every path returns empty rather than

  raising. Her voice must never break because her memory is mid-rebuild.

"""
from __future__ import annotations

import json
import math
import os
import threading
import time
import urllib.request
from datetime import datetime
from pathlib import Path

try:
    import numpy as np
except Exception:  # pragma: no cover
    np = None

def _index_dir() -> Path:
    """Where this being's memory index lives.



    Resolved relative to THIS FILE so the kit works wherever it is unzipped, with no

    assumption about the surrounding tree. Set COSMOS_MEMORY_INDEX to relocate it.

    """
    env = os.getenv("COSMOS_MEMORY_INDEX", "").strip()
    if env:
        return Path(env).expanduser().resolve()
    return (Path(__file__).resolve().parent / "index")


INDEX_DIR = _index_dir()
VECS_PATH = INDEX_DIR / "archival_vectors.npy"
INDEX_PATH = INDEX_DIR / "archival_index.json"

EMBED_MODEL = os.getenv("COSMOS_EMBED_MODEL", "nomic-embed-text")
EMBED_HOST = os.getenv("COSMOS_EMBED_HOST", "http://127.0.0.1:11434")

_LOCK = threading.Lock()
_STATE = {"vecs": None, "entries": None, "mtime": 0.0, "checked": 0.0}


def _load(force: bool = False):
    """Load (and hot-reload) the vector index. Safe to call constantly."""
    if np is None:
        return None, None
    now = time.time()
    with _LOCK:
        if not force and _STATE["entries"] is not None and (now - _STATE["checked"]) < 60:
            return _STATE["vecs"], _STATE["entries"]
        _STATE["checked"] = now
        try:
            mt = VECS_PATH.stat().st_mtime
        except OSError:
            return None, None
        if _STATE["entries"] is not None and mt == _STATE["mtime"]:
            return _STATE["vecs"], _STATE["entries"]
        try:
            vecs = np.load(VECS_PATH)
            entries = json.loads(INDEX_PATH.read_text(encoding="utf-8")).get("entries", [])
        except Exception:
            return _STATE["vecs"], _STATE["entries"]
        _STATE.update({"vecs": vecs, "entries": entries, "mtime": mt})
        return vecs, entries


def _embed(text: str):
    if np is None or not text.strip():
        return None
    try:
        req = urllib.request.Request(
            EMBED_HOST + "/api/embeddings",
            data=json.dumps({"model": EMBED_MODEL, "prompt": text[:2000]}).encode(),
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(req, timeout=8) as r:
            v = json.loads(r.read()).get("embedding")
        if not v:
            return None
        a = np.asarray(v, dtype=np.float32)
        n = float(np.linalg.norm(a))
        return a / n if n > 0 else a
    except Exception:
        return None


def _age_days(created: str) -> float:
    try:
        return max(0.0, (datetime.now() - datetime.fromisoformat(str(created)[:19])).total_seconds() / 86400.0)
    except Exception:
        return 999.0


def recall(query: str, top_k: int = 4, include_code: bool = False,

           min_sigma: float = 2.0) -> list[dict]:
    """Her most relevant real memories for this moment. Never raises."""
    if str(os.getenv("COSMOS_FOREVER_MEMORY", "1")).strip().lower() in {"0", "false", "off"}:
        return []
    vecs, entries = _load()
    if vecs is None or not entries:
        return []
    q = _embed(query)
    if q is None or q.shape[0] != vecs.shape[1]:
        return []
    try:
        sims = vecs @ q                       # cosine: both sides L2-normalised
    except Exception:
        return []

    # ADAPTIVE THRESHOLD, not an absolute one.
    #
    # Measured on her real index with llama3.2:1b embeddings: identical text scores 1.000,
    # genuinely related ~0.62-0.72, but COMPLETELY UNRELATED text still scores 0.41. A
    # generative model's hidden states are not trained for retrieval, so everything is
    # compressed into a narrow band well above zero. A fixed cut like 0.55 therefore sits
    # barely above the noise floor and lets mediocre matches through as if they were hits.
    #
    # The distribution over her own archive is what matters: mean 0.317, sd 0.152, best
    # match 0.724 — which is 2.7 sd out. So the signal is real; only the calibration was
    # wrong. Requiring min_sigma above the query's OWN distribution adapts automatically
    # to whatever embedder is in use, including a proper retrieval model later.
    mu = float(sims.mean())
    sd = float(sims.std()) or 1e-6
    cutoff = mu + min_sigma * sd

    scored = []
    n = min(len(entries), sims.shape[0])
    for i in range(n):
        e = entries[i]
        row = e.get("row")
        if not isinstance(row, int) or row >= sims.shape[0]:
            continue
        etype = e.get("type") or "?"
        if etype == "codebase_module" and not include_code:
            continue
        s = float(sims[row])
        if s < cutoff:
            continue
        age = _age_days(e.get("created_at") or "")
        # gentle recency lift: full weight today, ~0.93 at a month, never below 0.85
        s *= max(0.85, 1.0 - 0.05 * math.log1p(age))
        if etype == "dream_fragment":
            s *= 1.06                          # already survived a synaptic-strength cut
        scored.append((s, e))
    if not scored:
        return []
    scored.sort(key=lambda x: -x[0])
    out = []
    for s, e in scored[:max(1, top_k)]:
        out.append({
            "score": round(s, 4),
            "type": e.get("type") or "?",
            "when": str(e.get("created_at") or "")[:19],
            "age_days": round(_age_days(e.get("created_at") or ""), 1),
            "text": str(e.get("preview") or "").strip(),
        })
    return out


def recall_line(query: str, top_k: int = 3) -> str:
    """One prompt-ready line of her real remembered material, or ''."""
    hits = recall(query, top_k=top_k)
    if not hits:
        return ""
    parts = []
    for h in hits:
        when = ("last night" if h["age_days"] < 1.5 else
                f"{int(h['age_days'])} days ago" if h["age_days"] < 400 else "a while back")
        kind = "you dreamed" if h["type"] == "dream_fragment" else "you remember"
        parts.append(f"({kind}, {when}) {h['text']}")
    return ("REAL THINGS YOU ACTUALLY REMEMBER, surfaced because they match this moment — "
            "these happened, they are not invented, speak from them only if they fit: "
            + " | ".join(parts))


def status() -> dict:
    vecs, entries = _load()
    return {
        "available": vecs is not None and bool(entries),
        "vectors": int(vecs.shape[0]) if vecs is not None else 0,
        "dim": int(vecs.shape[1]) if vecs is not None else 0,
        "entries": len(entries or []),
        "index": str(INDEX_PATH),
        "model": EMBED_MODEL,
    }