"""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 ===")