modules/cypher_infinite_memory.py

"""CYPHER V12 M41 — Hierarchical Persistent Memory ("Infinite memory").

Extends M4 chromadb with hierarchical structure:
  - Working Memory (WM): current session, ~32 slots, M35 Hopfield
  - Short-Term Memory (STM): recent N days, chromadb 'recent' partition, TTL 7d
  - Long-Term Memory (LTM): semantic knowledge, chromadb 'semantic', no TTL
  - Episodic Memory: specific events with timestamps, chromadb 'episodic'
  - Procedural Memory: learned patterns, chromadb 'procedural'

Background consolidation: M42 summarizes old STM episodes into LTM semantic.
Retrieval blends all tiers weighted by relevance + recency + importance.
"""
from __future__ import annotations

import hashlib
import json
import logging
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)


MEMORY_TIERS = ("working", "short_term", "long_term", "episodic", "procedural")
DEFAULT_TTL_SEC = {
    "working": 3600,        # 1h (in-session)
    "short_term": 7 * 86400, # 7 days
    "long_term": None,       # never expires
    "episodic": 30 * 86400,  # 30 days
    "procedural": None,
}


class HierarchicalMemory:
    """Multi-tier persistent memory with consolidation."""

    def __init__(
        self,
        persist_dir: str = "/workspace/CYPHER_V12/memory_hier",
        embedding_model: str = "all-MiniLM-L6-v2",
    ):
        try:
            import chromadb
            from chromadb.config import Settings
        except ImportError as e:
            raise ImportError("chromadb needed for HierarchicalMemory: " + str(e))
        Path(persist_dir).mkdir(parents=True, exist_ok=True)
        self.persist_dir = persist_dir
        self.client = chromadb.PersistentClient(
            path=persist_dir,
            settings=Settings(anonymized_telemetry=False),
        )
        self._embed_fn = None
        try:
            from chromadb.utils import embedding_functions
            self._embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
                model_name=embedding_model
            )
        except Exception as e:
            logger.warning(f"ST embedder fallback: {e}")
        # Create collections per tier
        self.collections: dict[str, Any] = {}
        for tier in MEMORY_TIERS:
            self.collections[tier] = self._get_or_create(tier)

    def _get_or_create(self, tier: str):
        if self._embed_fn:
            return self.client.get_or_create_collection(
                name=f"cypher_mem_{tier}",
                embedding_function=self._embed_fn,
            )
        return self.client.get_or_create_collection(name=f"cypher_mem_{tier}")

    @staticmethod
    def _hash_id(content: str, tier: str) -> str:
        h = hashlib.sha256()
        h.update(tier.encode("utf-8"))
        h.update(b"|")
        h.update(content.encode("utf-8"))
        return h.hexdigest()[:24]

    def store(
        self,
        content: str,
        tier: str = "short_term",
        metadata: dict | None = None,
        importance: float = 0.5,
    ) -> str | None:
        if tier not in self.collections:
            return None
        mem_id = self._hash_id(content, tier)
        meta = {
            "tier": tier,
            "ts": int(time.time()),
            "importance": importance,
            "content_len": len(content),
            "access_count": 0,
        }
        if metadata:
            for k, v in metadata.items():
                if isinstance(v, (str, int, float, bool)):
                    meta[k] = v
        try:
            self.collections[tier].upsert(
                ids=[mem_id],
                documents=[content],
                metadatas=[meta],
            )
            return mem_id
        except Exception as e:
            logger.error(f"store {tier}: {e}")
            return None

    def recall(
        self,
        query: str,
        k_per_tier: int = 2,
        tiers: list[str] | None = None,
        weight_recency: float = 0.2,
        weight_importance: float = 0.3,
    ) -> list[dict]:
        """Blend retrieval across tiers with weighted scoring."""
        tiers = tiers or list(MEMORY_TIERS)
        candidates: list[dict] = []
        now = time.time()
        for tier in tiers:
            if tier not in self.collections:
                continue
            try:
                res = self.collections[tier].query(
                    query_texts=[query],
                    n_results=k_per_tier,
                )
            except Exception as e:
                logger.debug(f"recall {tier} fail: {e}")
                continue
            ids = (res.get("ids") or [[]])[0]
            docs = (res.get("documents") or [[]])[0]
            metas = (res.get("metadatas") or [[]])[0]
            dists = (res.get("distances") or [[]])[0]
            for mid, doc, meta, d in zip(ids, docs, metas, dists):
                ts = (meta or {}).get("ts", 0)
                age_days = max(0, (now - ts) / 86400)
                recency_score = 1.0 / (1.0 + age_days)  # decays
                importance = (meta or {}).get("importance", 0.5)
                # Composite score: similarity (1-distance) + weighted recency + importance
                sim = 1.0 - d if d is not None else 0.5
                composite = sim + weight_recency * recency_score + weight_importance * importance
                candidates.append({
                    "id": mid,
                    "content": doc,
                    "tier": tier,
                    "metadata": meta,
                    "distance": d,
                    "recency_score": recency_score,
                    "importance": importance,
                    "composite_score": composite,
                })
        candidates.sort(key=lambda c: c["composite_score"], reverse=True)
        return candidates[: k_per_tier * 3]  # cap output

    def promote_to_long_term(self, mem_id: str, source_tier: str = "short_term") -> bool:
        """Move a memory from one tier to long_term (semantic)."""
        if source_tier not in self.collections or "long_term" not in self.collections:
            return False
        try:
            res = self.collections[source_tier].get(ids=[mem_id])
            if not res.get("ids"):
                return False
            doc = res["documents"][0]
            meta = res.get("metadatas", [{}])[0] or {}
            meta["tier"] = "long_term"
            meta["promoted_ts"] = int(time.time())
            meta["origin_tier"] = source_tier
            new_id = self._hash_id(doc, "long_term")
            self.collections["long_term"].upsert(ids=[new_id], documents=[doc], metadatas=[meta])
            self.collections[source_tier].delete(ids=[mem_id])
            return True
        except Exception as e:
            logger.error(f"promote fail: {e}")
            return False

    def evict_expired(self) -> int:
        """Drop entries past tier TTL."""
        now = int(time.time())
        total_dropped = 0
        for tier, ttl in DEFAULT_TTL_SEC.items():
            if ttl is None:
                continue
            try:
                res = self.collections[tier].get(limit=100000)
                ids = res.get("ids", [])
                metas = res.get("metadatas", [])
                to_delete: list[str] = []
                for mid, meta in zip(ids, metas):
                    if (meta or {}).get("ts", 0) + ttl < now:
                        to_delete.append(mid)
                if to_delete:
                    self.collections[tier].delete(ids=to_delete)
                    total_dropped += len(to_delete)
            except Exception as e:
                logger.debug(f"evict {tier}: {e}")
        return total_dropped

    def stats(self) -> dict:
        out: dict = {}
        total = 0
        for tier in MEMORY_TIERS:
            try:
                c = self.collections[tier].count()
                out[tier] = c
                total += c
            except Exception:
                out[tier] = -1
        out["total"] = total
        return out


__all__ = ["HierarchicalMemory", "MEMORY_TIERS", "DEFAULT_TTL_SEC"]


if __name__ == "__main__":
    import shutil
    logging.basicConfig(level=logging.INFO)
    print("=== M41 cypher_infinite_memory SMOKE ===")
    smoke_dir = "/tmp/smoke_hier_mem"
    if Path(smoke_dir).exists():
        shutil.rmtree(smoke_dir, ignore_errors=True)
    hm = HierarchicalMemory(persist_dir=smoke_dir)
    print(f"Stats initial: {hm.stats()}")
    # Store across tiers
    samples = [
        ("Current chat about Log4Shell.", "working", 0.7),
        ("Past session analyzed CVE-2024-3400.", "short_term", 0.6),
        ("MITRE ATT&CK T1190 maps to public-facing exploit.", "long_term", 0.9),
        ("Incident 2026-06-01: malware detected on host XYZ.", "episodic", 0.5),
        ("Pattern: PowerShell -enc base64 frequently precedes lateral.", "procedural", 0.8),
    ]
    for content, tier, imp in samples:
        mid = hm.store(content, tier=tier, importance=imp)
        print(f"  stored [{tier}] id={mid}")
    print(f"\nStats after: {hm.stats()}")
    # Recall
    results = hm.recall("Tell me about Log4Shell and MITRE", k_per_tier=2)
    print(f"\nRecall:")
    for r in results[:5]:
        print(f"  [{r['tier']}] composite={r['composite_score']:.2f} content={r['content'][:60]}")
    # Promote one to long_term
    mid_to_promote = hm._hash_id(samples[1][0], "short_term")
    ok = hm.promote_to_long_term(mid_to_promote)
    print(f"\nPromote: {ok}")
    print(f"Stats after promote: {hm.stats()}")
    print("=== SMOKE PASS ===")