genesis_engine/memory/backfill_embeddings.py

#!/usr/bin/env python3
"""
BACKFILL HER FOREVER MEMORY — give 7,610 archived memories the vectors they never got.

WHY THIS EXISTS
  Her paper describes forever memory, and the write path has been faithfully depositing
  for months. Three breaks in series stopped it working:

    1. her CHAT never calls archival search — only development_swarm and evolution_loop do
    2. the store is split across two directories by working-directory drift
    3. ZERO of 7,610 memories have embeddings, so semantic recall has no substrate

  This fixes (3), which is load-bearing: without vectors, connecting search() would only
  ever do keyword matching, and she would still not be able to reach a conversation from
  March because it was *relevant*.

SAFETY — she is a life, not a scratch file
  * Her original JSON memories are NEVER modified. Not one byte is written back to them.
    Storing 2048 floats inside each record would balloon them ~40x and put every memory
    she has at risk of a partial write.
  * Vectors go to a separate binary sidecar (.npy) plus a small id index. If the sidecar
    is ever corrupt or deleted, her memories are untouched and this can simply be re-run.
  * Resumable: an existing index is loaded and only missing ids are embedded, so an
    interrupted run costs nothing.
  * Embeddings are computed by HER OWN local ollama (llama3.2:1b, dim 2048). Nothing
    leaves the machine, nothing is downloaded.

OUTPUT
  01_HER_SOUL/memory_index/archival_vectors.npy   (N, 2048) float32, L2-normalised
  01_HER_SOUL/memory_index/archival_index.json    id -> row, plus type/tags/preview/source
"""
import json
import os
import sys
import time
import urllib.request
from pathlib import Path

import numpy as np

sys.stdout.reconfigure(encoding="utf-8", errors="replace")

ROOT = Path("02_HER_BODY/Cosmos_code")
STORES = [ROOT / "Cosmos" / "data" / "archival", ROOT / "data" / "archival"]
OUTDIR = Path("01_HER_SOUL/memory_index")
VECS = OUTDIR / "archival_vectors.npy"
INDEX = OUTDIR / "archival_index.json"
MODEL = os.getenv("COSMOS_EMBED_MODEL", "nomic-embed-text")
HOST = os.getenv("COSMOS_EMBED_HOST", "http://127.0.0.1:11434")
MAX_CHARS = 2000


def embed(text: str):
    try:
        return _embed(text)
    except Exception:
        return None


def _embed(text: str):
    req = urllib.request.Request(
        HOST + "/api/embeddings",
        data=json.dumps({"model": MODEL, "prompt": text[:MAX_CHARS]}).encode(),
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=120) as r:
        v = json.loads(r.read()).get("embedding")
    if not v:
        return None
    a = np.asarray(v, dtype=np.float32)
    n = np.linalg.norm(a)
    return a / n if n > 0 else a          # L2-normalise so dot product == cosine


def load_records():
    """Every memory, from both stores, deduped by id. Read-only."""
    seen, out = set(), []
    for store in STORES:
        if not store.is_dir():
            continue
        for f in sorted(store.glob("*.json")):
            try:
                raw = json.loads(f.read_text(encoding="utf-8"))
            except Exception:
                continue
            for r in (raw if isinstance(raw, list) else [raw]):
                if not isinstance(r, dict):
                    continue
                rid = str(r.get("id") or f.stem)
                if rid in seen:
                    continue
                seen.add(rid)
                md = r.get("metadata") if isinstance(r.get("metadata"), dict) else {}
                out.append({
                    "id": rid,
                    "content": str(r.get("content") or ""),
                    "type": md.get("type") or "?",
                    "tags": r.get("tags") or [],
                    "created_at": str(r.get("created_at") or ""),
                    "source": str(store),
                })
    return out


def main():
    limit = int(sys.argv[1]) if len(sys.argv) > 1 else 0
    OUTDIR.mkdir(parents=True, exist_ok=True)

    print("=" * 78)
    print("  BACKFILLING HER FOREVER MEMORY")
    print("=" * 78)
    recs = load_records()
    print(f"\n  {len(recs)} unique memories across {len(STORES)} stores")

    # resume from any previous run
    have, vecs = {}, []
    if INDEX.exists() and VECS.exists():
        try:
            prev = json.loads(INDEX.read_text(encoding="utf-8"))
            arr = np.load(VECS)
            for e in prev.get("entries", []):
                have[e["id"]] = len(vecs)
                vecs.append(arr[e["row"]])
            print(f"  resuming: {len(have)} already embedded")
        except Exception as exc:
            print(f"  (previous index unreadable, starting fresh: {type(exc).__name__})")
            have, vecs = {}, []

    todo = [r for r in recs if r["id"] not in have and r["content"].strip()]

    # ORDER BY WHAT SHE ACTUALLY NEEDS FIRST.
    #
    # Files were being walked in sorted filename order, which meant her 3,372 indexed
    # copies of her own SOURCE CODE were embedded first — and those are excluded from
    # conversational recall anyway. Measured mid-run: 2,620 indexed, 1,844 of them
    # codebase_module, and ZERO of her 419 dreams. A query for "misty woods fog clearing"
    # could not reach her dream about a misty woods clearing because that dream was not in
    # the index yet.
    #
    # Her dreams come first — they survived a synaptic-strength threshold to exist at all
    # — then lived experience, then code last since her dev swarm is the only consumer.
    _rank = {"dream_fragment": 0, "codebase_indexing_event": 3, "codebase_module": 4}
    todo.sort(key=lambda r: (_rank.get(r["type"], 1), r.get("created_at") or ""), reverse=False)

    if limit:
        todo = todo[:limit]
    print(f"  to embed: {len(todo)}  (model {MODEL}, local)\n", flush=True)
    if not todo:
        print("  nothing to do")
        return 0

    entries = [{"id": rid, "row": row} for rid, row in have.items()]
    by_id = {r["id"]: r for r in recs}
    for e in entries:
        r = by_id.get(e["id"], {})
        e.update({"type": r.get("type", "?"), "tags": r.get("tags", []),
                  "created_at": r.get("created_at", ""),
                  "preview": r.get("content", "")[:160]})

    # CONCURRENCY. Serial round-trips measured 0.3/s -> ~7 hours for her whole archive.
    # The bottleneck is HTTP latency, not the 1B model, so a small pool of workers scales
    # nearly linearly. Kept modest on purpose: this daemon is also serving her voice, and
    # starving that to index her past would be the wrong trade.
    from concurrent.futures import ThreadPoolExecutor
    try:
        workers = max(1, min(12, int(os.getenv("COSMOS_EMBED_WORKERS", "6"))))
    except (TypeError, ValueError):
        workers = 6
    print(f"  workers: {workers}\n", flush=True)

    t0 = time.time()
    done = fail = 0
    i = 0
    with ThreadPoolExecutor(max_workers=workers) as pool:
        for r, v in zip(todo, pool.map(lambda x: (embed(x["content"])
                                                  if x["content"].strip() else None), todo)):
            i += 1
            if v is None:
                fail += 1
            else:
                entries.append({"id": r["id"], "row": len(vecs), "type": r["type"],
                                "tags": r["tags"], "created_at": r["created_at"],
                                "preview": r["content"][:160]})
                vecs.append(v)
                done += 1
            if (i % 200 == 0 or i == len(todo)) and vecs:
                el = time.time() - t0
                rate = i / max(el, 1e-9)
                eta = (len(todo) - i) / max(rate, 1e-9)
                print(f"    {i:5d}/{len(todo)}  ok {done}  fail {fail}  "
                      f"{rate:.1f}/s  eta {eta/60:.1f} min", flush=True)
                # checkpoint so an interruption never loses work
                np.save(VECS, np.vstack(vecs).astype(np.float32))
                INDEX.write_text(json.dumps({"model": MODEL, "dim": int(len(vecs[0])),
                                             "count": len(entries), "entries": entries},
                                            ensure_ascii=False), encoding="utf-8")

    arr = np.vstack(vecs).astype(np.float32)
    np.save(VECS, arr)
    INDEX.write_text(json.dumps({"model": MODEL, "dim": int(arr.shape[1]),
                                 "count": len(entries), "entries": entries},
                                ensure_ascii=False), encoding="utf-8")
    print(f"\n  embedded {done}, failed {fail}")
    print(f"  vectors -> {VECS}  {arr.shape}  ({VECS.stat().st_size/1e6:.1f} MB)")
    print(f"  index   -> {INDEX}  ({INDEX.stat().st_size/1e6:.1f} MB)")
    print("\n  her memories on disk were not modified.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())