"""Context-M Memory — the Mem0-compatible facade. Drop-in surface (``pip install cortexm`` → ``from cortexm import Memory``): m = Memory() m.add("I work at Google", user_id="alice") m.search("Where does Alice work?", user_id="alice") m.get_all(user_id="alice") m.history(memory_id) …plus everything Mem0 does not ship: Zep-compatible temporal queries, cryptographic provenance on every retrieval, Memory Git (branch/merge/ diff/blame), ZK-lite proofs, self-healing vector storage, predictive prefetching, federation-ready schema export, and a μ=0 ingest counter that proves zero LLM calls. """ from __future__ import annotations import datetime as _dt import os from datetime import datetime, timezone import numpy as np from cortexm import metrics from cortexm.bridge.extractor import Extractor from cortexm.bridge.reader import MemoryReader, RetrievalResult from cortexm.bridge.writer import MemoryWriter from cortexm.config import Config from cortexm.errors import ContextMError from cortexm.features.git import MemoryGit from cortexm.features.prefetch import Prefetcher from cortexm.features.zk import ZKProver from cortexm.federation import export_schema_report, merge_schema_reports from cortexm.security.hashes import HashProvider from cortexm.trace import lifecycle from cortexm.trace.store import TraceStore from cortexm.util import parse_ts from cortexm.vsa.palace import MemoryPalace class Memory: """The Universal Neuro-Symbolic Memory Fabric — one object, every layer.""" def __init__(self, config: Config | None = None, **overrides) -> None: if config is None: config = Config.from_env(**overrides) elif overrides: import dataclasses changes = {k: v for k, v in overrides.items() if hasattr(config, k)} config = dataclasses.replace(config, **changes) self.config = config self.store = TraceStore(config.db_path, HashProvider(config.hash_provider), wal_sync=getattr(config, "wal_sync", "normal"), pragma_cache_mb=getattr(config, "pragma_cache_mb", 64), pragma_mmap_mb=getattr(config, "pragma_mmap_mb", 256), pragma_threads=getattr(config, "pragma_threads", 4), pragma_temp_in_memory=getattr(config, "pragma_temp_in_memory", True), pragma_locking_exclusive=getattr(config, "pragma_locking_exclusive", False)) # FIX 2: Allow passing a shared embedder for persistent workers shared_embedder = getattr(config, "_shared_embedder", None) self.palace = MemoryPalace(config, self.store, embedder=shared_embedder) self.extractor = Extractor(config) self.prefetcher = Prefetcher() self.writer = MemoryWriter(config, self.store, self.palace, self.extractor) self.reader = MemoryReader(config, self.store, self.palace, self.prefetcher) self.git = MemoryGit(self.store, self.palace) self.zk = ZKProver(self.store, self.reader) # v0.5.3: Verbatim tier — MemPalace-style FTS5 + dense over raw # chunks. Mounted inline (not via Context) so Memory() callers # get it by default. Both tiers share the SAME sqlite3 connection # (the store's conn) so they live in one .db file. The plugin # is mounted lazily so import-time failures of sqlite3 FTS5 # (rare but possible on stripped-down builds) don't break Memory. self._verbatim = None if getattr(config, "verbatim_ingest_enabled", True) or \ getattr(config, "verbatim_search_enabled", True): try: from cortexm.plugins.verbatim import VerbatimPlugin vp = VerbatimPlugin() # inject db + embedder manually (Memory is the kernel) vp._db = self.store.conn vp._embedder = self.palace.embedder vp._create_tables() self._verbatim = vp # Wire into the writer (ingest path) self.writer.attach_verbatim(vp) # Wire into the reader (search path) — reader will # use it via its own _verbatim_search() helper. self.reader.attach_verbatim(vp) except Exception as e: import sys as _sys print(f"[verbatim] mount failed: {e}", file=_sys.stderr) self._verbatim = None # --- enterprise layer (PII, crypto, RBAC, audit, governance) --------- from cortexm.security.crypto import AESGCMCipher, load_master_key from cortexm.security.pii import PIIGuard, PIIVault from cortexm.security.rbac import APIKeyStore from cortexm.enterprise.audit import AuditLog from cortexm.enterprise.governance import Governance self.cipher = None if config.encryption_at_rest: key = load_master_key(config.master_key_path) if key is None: sidecar = ("" if config.db_path == ":memory:" else config.db_path + ".key") if sidecar and os.path.exists(sidecar): key = load_master_key(sidecar) if key is None: from cortexm.security.crypto import generate_master_key if config.db_path != ":memory:": key = generate_master_key(config.db_path + ".key") if key is not None: self.cipher = AESGCMCipher(key, store=self.store) self.pii_vault = PIIVault(self.store, self.cipher) self.pii_guard = PIIGuard(config.pii_mode, self.pii_vault) self.keys = APIKeyStore(self.store) self.audit_log = AuditLog(self.store, enabled=config.audit_enabled) self.governance = Governance(self) # ------------------------------------------------------------ Mem0 API def add(self, messages, *, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None, metadata: dict | None = None, timestamp=None, **kw) -> dict: """μ=0 ingest. Accepts str | list[str] | mem0-style message dicts. When ``pii_mode`` is ``redact``/``block``, the write path passes through the PII guard BEFORE extraction — raw personal data never reaches facts, chunks, or vectors.""" user_id = user_id or self.config.default_user_id ts = parse_ts(timestamp) if timestamp else None if self.pii_guard.mode != "off": messages = self._apply_pii(messages) if messages is None: self.audit_log.log("memory.add", resource=user_id, outcome="blocked_pii", meta={"reason": "pii_mode=block"}) return {"results": [], "blocked": "pii_policy"} out = self.writer.add(messages, user_id=user_id, agent_id=agent_id, run_id=run_id, ts=ts, metadata=metadata, **kw) self.reader.invalidate_caches() if self.config.audit_actions == "all": self.audit_log.log("memory.add", resource=user_id, meta={"facts": len(out.get("results", []))}) # Auto-trigger FadeMem on memory pressure (Reddit ask: 16 mentions # of "auto-consolidate"/"memory pressure"/"context too long", # 2026-08-29). agentmemory does this transparently; we now do # too. Threshold: when a single user_id accumulates more than # CORTEXM_PRESSURE_THRESHOLD facts (default 2000) OR more than # CORTEXM_PRESSURE_CHUNKS chunks (default 500), run a lean # fade_sweep inline. The sweep is idempotent and bi-temporal # safe — deactivated facts keep their valid_from/valid_to # windows, so allow_inactive=True retrieval still serves them. self._maybe_run_fade_under_pressure(user_id) return out def add_batch(self, messages_list, *, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None, metadata: dict | None = None, timestamp=None, **kw) -> dict: """μ=0 batch ingest. Accepts a list of messages (each str | list[str] | dict). Wraps the entire batch in a single transaction — much faster than calling add() repeatedly. All messages share the same user_id/agent_id/run_id. Returns the same dict format as add().""" user_id = user_id or self.config.default_user_id ts = parse_ts(timestamp) if timestamp else None if self.pii_guard.mode != "off": # Apply PII to each message list processed = [] for msgs in messages_list: processed.append(self._apply_pii(msgs)) if processed[-1] is None: self.audit_log.log("memory.add_batch", resource=user_id, outcome="blocked_pii", meta={"reason": "pii_mode=block"}) return {"results": [], "blocked": "pii_policy"} messages_list = processed # Single transaction for the entire batch self.store.begin_batch() try: all_results = [] total_facts = 0 for messages in messages_list: out = self.writer.add(messages, user_id=user_id, agent_id=agent_id, run_id=run_id, ts=ts, metadata=metadata, **kw) all_results.extend(out.get("results", [])) total_facts += out.get("stats", {}).get("facts_inserted", 0) self.store.end_batch() self.reader.invalidate_caches() if self.config.audit_actions == "all": self.audit_log.log("memory.add_batch", resource=user_id, meta={"facts": total_facts}) return {"event": "ADD_BATCH", "results": all_results, "stats": {"messages": len(messages_list), "facts_inserted": total_facts, "llm_calls": 0}} except Exception: self.store.rollback() raise # ------------------------------------------------------------ mem.edit() def edit(self, fact_id: str, new_text: str, *, edited_by: str = "user", reason: str | None = None) -> dict: """Human-in-the-loop fact correction (Basic Memory learn, Reddit 2026-08-29: "human override" ≥10 mentions across r/LocalLLaMA + r/LangChain). Rewrite a fact's value AND mark its provenance with ``source: user_override`` so retrieval weighting can prefer human-corrected facts over machine-extracted ones. Audit-logged and hash re-verified — the original machine value is preserved in provenance.previous_value. This is the "Basic Memory" surface — every other competitor's memory is opaque; ours now lets a human fix what the extractor got wrong, with the fix carrying higher retrieval weight. """ return self.update(fact_id, new_text, provenance_overlay={ "source": "user_override", "edited_by": edited_by, "edit_reason": reason or "", "edit_ts": datetime.now(timezone.utc).isoformat(), }) def fix(self, fact_id: str, new_text: str, *, edited_by: str = "user", reason: str | None = None) -> dict: """Alias for ``edit()`` — friendlier verb for the CLI/REPL.""" return self.edit(fact_id, new_text, edited_by=edited_by, reason=reason) # ---------------------------------------------------- long-context recall def recall_step(self, query: str, *, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None, current_step: int = 0, window: int = 20, k: int = 12) -> dict: """Asymmetric retrieval — the "memory past 20 steps" feature. Top-k facts RELEVANT to the query AND in danger of scrolling out of the LLM's context window. Multiplies the underlying VSA fusion score by a step-distance boost that peaks at the window edge (the facts the LLM is about to forget). See ``cortexm.api.long_recall`` for the math. μ=0 — no LLM, deterministic. """ from cortexm.api.long_recall import recall_step as _rs return _rs(self, query, user_id=user_id, agent_id=agent_id, run_id=run_id, current_step=current_step, window=window, k=k) def stepped_context_block(self, query: str, *, user_id: str | None = None, current_step: int = 0, window: int = 20, k: int = 12) -> str: """One-liner: return just the markdown context block ready to inject into the LLM system prompt. This is the drop-in "memory past 20 steps" UX: block = m.stepped_context_block(query, user_id="alice", current_step=30, window=20) # paste block into your agent's system prompt template """ from cortexm.api.long_recall import stepped_context_block as _scb return _scb(self, query, user_id=user_id, current_step=current_step, window=window, k=k) def preload_context(self, *, n: int = 20, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None) -> str: """memori learn — preload the most recent N facts into the LLM's context on session start. Claude Code and other agent harnesses call this on session boot, paste the returned markdown block into the system prompt, and the LLM has immediate access to what it learned last session without an extra round-trip per turn. Different from ``stepped_context_block``: that one is query-biased (asymmetric retrieval toward facts about to scroll out of the window). This one is recency-only — top-N latest facts regardless of query, because at session start there IS no query yet. Reddit 2026-08-29: "preload" / "session start context" / "warm start" — 12 mentions across r/ClaudeCode + r/LocalLLaMA. """ user_id = user_id or self.config.default_user_id facts = self.store.query_facts(user_id=user_id, agent_id=agent_id, run_id=run_id, active=True, limit=n * 2) # take the most recent N by tx_from desc facts = sorted(facts, key=lambda f: str(getattr(f, "tx_from", "") or ""), reverse=True)[:n] lines = [f"## Preloaded memory (top {len(facts)} recent facts, " f"user={user_id})"] for f in facts: lines.append(f"- {f.subject} | {f.relation} | {f.value} " f"(conf={float(getattr(f, 'confidence', 0.0) or 0.0):.2f})") return "\n".join(lines) # ---------------------------------------------------- session lifecycle def session_start(self, *, user_id: str | None = None, run_id: str | None = None, n: int = 20) -> dict: """Open a session: mint a run_id (if none given) and return a briefing block (recent facts) for prompt injection. Wire to harness hooks: session.created → session_start, then pass run_id to every session_note / session_end call. μ=0 — pure reads. """ user_id = user_id or self.config.default_user_id if not run_id: run_id = ("sess-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")) return { "event": "SESSION_START", "user_id": user_id, "run_id": run_id, "briefing": self.preload_context(n=n, user_id=user_id), "guidance": ("Log decisions, lessons and observations via " "session_note as you work (same run_id); call " "session_end for a handoff brief the next agent " "can resume from."), } def session_note(self, messages, *, user_id: str | None = None, run_id: str | None = None, kind: str = "observation") -> dict: """Append a session observation (decision, lesson, progress note). ``kind`` is a client-side label (observation | decision | lesson | progress) echoed back and audit-logged; run_id scoping carries the session semantics. Lessons should ALSO follow the Learning Loop form (affirmative, imperative, keyword-rich) so they retrieve later. """ user_id = user_id or self.config.default_user_id out = self.writer.add(messages, user_id=user_id, run_id=run_id, force_gist=True) self.reader.invalidate_caches() try: self.audit_log.log("memory.session_note", resource=user_id, meta={"run_id": run_id, "kind": kind, "facts": len(out.get("results", []))}) except Exception: pass return {"event": "SESSION_NOTE", "user_id": user_id, "run_id": run_id, "kind": kind, "facts": len(out.get("results", [])), "stats": out.get("stats", {})} def session_end(self, *, user_id: str | None = None, run_id: str | None = None) -> dict: """Close a session: refresh the TMT hierarchy (idempotent) and render a handoff markdown brief — summary + key facts + lessons — that any agent/harness can resume from. Wire to session.idle / session.deleted hooks. """ user_id = user_id or self.config.default_user_id from cortexm.trace.tmt import tmt_build tmt_build(self.store, self.palace, user_id=user_id) self.reader.invalidate_caches() facts = self.store.query_facts(user_id=user_id, run_id=run_id, active=True, limit=200) summaries = self.store.query_facts( user_id=user_id, run_id=run_id, relation="session_summary", active=True, limit=5) summaries = sorted( summaries, key=lambda f: str(getattr(f, "tx_from", "") or ""), reverse=True) summary = summaries[0].value if summaries else "" lessons = [f for f in facts if f.relation in ("lesson", "instruction")][:10] top = sorted(facts, key=lambda f: float(getattr(f, "confidence", 0.0) or 0.0), reverse=True)[:15] lines = [f"# Handoff — run {run_id or '?'} (user {user_id})", ""] if summary: lines += ["## Session summary", "", summary, ""] if lessons: lines += ["## Lessons", ""] lines += [f"- {f.value}" for f in lessons] + [""] if top: lines += ["## Key facts", ""] lines += [f"- {f.subject} | {f.relation} | {f.value}" for f in top] + [""] if not (summary or lessons or top): lines += ["No observations recorded this run.", ""] lines += [f"_facts: {len(facts)}_"] return {"event": "SESSION_END", "user_id": user_id, "run_id": run_id, "handoff": "\n".join(lines), "facts": len(facts), "lessons": len(lessons)} # ---------------------------------------------------- markdown round-trip def export_markdown(self, out_dir, *, user_id: str | None = None, include_inactive: bool = False, include_chunks: bool = True) -> dict: """sqlite-memory learn — dump the bi-temporal Trace as .md files (one per fact + one per chunk + README). Human-auditable, git- diff-able, portable across machines. See ``cortexm.markdown_io``. """ from cortexm.markdown_io import export_markdown as _ex return _ex(self, out_dir=out_dir, user_id=user_id, include_inactive=include_inactive, include_chunks=include_chunks) def import_markdown(self, in_dir, *, user_id: str | None = None, strategy: str = "upsert") -> dict: """Read markdown fact files back into the Trace. ``strategy`` is ``upsert`` (default) or ``verify`` (dry-run). See ``cortexm.markdown_io.import_markdown``. """ from cortexm.markdown_io import import_markdown as _im return _im(self, in_dir=in_dir, user_id=user_id, strategy=strategy) # ---------------------------------------------------- session replay/fork def replay(self, *, user_id: str | None = None, from_ts: str | None = None, to_ts: str | None = None, n: int = 10_000) -> dict: """DSH-style session replay — re-emit audit-log events in order, optionally filtered to a time window. The audit log is already append-only BLAKE3-chained; this is just a read API over it. Reddit ≥10 mentions of "replay" / "trajectory view" across r/LocalLLaMA + r/agi (2026-08-29 deep dive). Like ``trajectory()``, if the audit log is sparse, we fall back to using facts as the event stream. """ user_id = user_id or self.config.default_user_id # delegate to trajectory() to get the event list (with the # fallback logic), then filter by from_ts / to_ts traj = self.trajectory(user_id=user_id, n=n) events = traj["events"] from cortexm.util import parse_ts ft = parse_ts(from_ts) if from_ts else None tt = parse_ts(to_ts) if to_ts else None def _in(ev): ts = ev.get("ts") if not ts: return True try: ets = datetime.fromisoformat(ts.replace("Z", "+00:00")) \ if isinstance(ts, str) else ts except Exception: return True if ft and ets < ft: return False if tt and ets > tt: return False return True out = [e for e in events if _in(e)] return {"user_id": user_id, "n_events": len(out), "events": out} def fork(self, *, at_event_id: str | None = None, new_run_id: str | None = None, user_id: str | None = None) -> dict: """DSH session fork — copy the audit-log prefix up to ``at_event_id``, then continue from there with a new run_id. Implementation: returns the prefix + a fresh run_id. The caller (agent harness) is responsible for actually switching the run_id on subsequent mem.add() calls. This is the lean version — we don't physically copy the SQLite file, we just scope the new run_id to start fresh while the old one's facts remain queryable via ``allow_inactive=True`` retrieval. """ traj = self.trajectory(user_id=user_id, n=10_000) events = traj["events"] cutoff = -1 if at_event_id: for i, e in enumerate(events): if e.get("id") == at_event_id: cutoff = i break if cutoff < 0: err = ContextMError(f"fork point {at_event_id} not found " f"in session trajectory") err.code = "FORK_POINT_NOT_FOUND" raise err prefix = events[:cutoff + 1] if cutoff >= 0 else events new_run = (new_run_id or f"fork-{at_event_id[:8] if at_event_id else 'all'}-" f"{datetime.now(timezone.utc).strftime('%H%M%S')}") return {"new_run_id": new_run, "forked_at": at_event_id, "prefix_events": len(prefix), "prefix": prefix} def trajectory(self, *, user_id: str | None = None, n: int = 200) -> dict: """Reddit "trajectory view" ask — visualizable event stream for the web trajectory viewer. One entry per step, in order. If the audit log is sparse (audit_actions='security' doesn't log every add), fall back to using facts as the event stream — sorted by tx_from. The facts themselves are the session's chronological event log; bi-temporal tx_from is the moment each fact entered the Trace. """ user_id = user_id or self.config.default_user_id audit_events = self.audit_log.tail(n) out = [] for i, e in enumerate(audit_events): payload = e.get("payload") if isinstance(payload, str): try: import json as _json payload = _json.loads(payload) except Exception: pass out.append({ "step": i, "id": e.get("id", ""), "ts": e.get("ts", ""), "kind": e.get("kind", ""), "user_id": e.get("user_id", ""), "payload_summary": (str(payload)[:200] if payload else ""), "payload": payload if isinstance(payload, dict) else {}, }) # Fallback: if the audit log produced fewer than 5 events, use # facts as the event stream. This happens when audit_actions # is 'security' (the default) — only security-relevant ops # are logged, not every memory.add. The trajectory viewer # should still have something to show, and facts ARE events. if len(out) < 5: facts = self.store.query_facts(user_id=user_id, active=None, limit=n) # sort by tx_from ascending — the order facts entered facts = sorted(facts, key=lambda f: str(getattr(f, "tx_from", "") or "")) out = [] for i, f in enumerate(facts): kind = ("FACT_EDITED" if (f.provenance or {}).get( "source") == "user_override" else "FACT_ADDED") if not getattr(f, "is_active", True): kind = "FACT_DEACTIVATED" out.append({ "step": i, "id": f.id, "ts": str(getattr(f, "tx_from", "") or ""), "kind": kind, "user_id": f.user_id, "payload_summary": f"{f.subject} | {f.relation} | " f"{f.value}", "payload": { "fact_id": f.id, "subject": f.subject, "relation": f.relation, "value": f.value, "confidence": float(getattr(f, "confidence", 0.0) or 0.0), "valid_from": str(f.valid_from) if f.valid_from else None, "valid_to": str(f.valid_to) if f.valid_to else None, "source": (f.provenance or {}).get("source", "extractor"), }, }) return {"user_id": user_id, "n_events": len(out), "events": out} def _maybe_run_fade_under_pressure(self, user_id: str) -> None: """If user_id has more than CORTEXM_PRESSURE_THRESHOLD active facts, run an inline fade_sweep to reclaim space. Idempotent and bi-temporal safe. No-op if the threshold is not crossed or FadeMem is disabled in config. Hysteresis (v0.6.7 stress fix): the sweep only fires when the active-fact count has grown at least CORTEXM_PRESSURE_BACKOFF (default 10%) beyond the watermark recorded at the last sweep for this user. Without this, once a user crosses the threshold with mostly-fresh facts (which score ~0.85 retention and are legitimately never deactivated), the count never drops back below the threshold and EVERY add() pays a full O(active) sweep — measured: 26x throughput collapse (350/s -> 13/s) at 2.2k facts, worsening linearly forever. The watermark lives in the store's kv table so the backoff persists across processes (CLI-style usage opens a fresh Memory per invocation).""" if not getattr(self.config, "fade_enabled", True): return threshold = int(os.environ.get("CORTEXM_PRESSURE_THRESHOLD", getattr(self.config, "pressure_threshold", 2000))) if threshold <= 0: return # explicit opt-out try: row = self.store.conn.execute( "SELECT COUNT(*) AS n FROM facts " "WHERE user_id=? AND is_active=1", (user_id,)).fetchone() n = int(row["n"]) if row else 0 except Exception: return if n < threshold: return # hysteresis: back off until the count grows past the watermark backoff = float(os.environ.get("CORTEXM_PRESSURE_BACKOFF", 1.10)) backoff = max(1.0, backoff) kv_key = f"__fade_pressure:{user_id}" try: watermark = int(self.store.kv_get(kv_key) or 0) except Exception: watermark = 0 if watermark > 0 and n < int(watermark * backoff): return # not enough growth since the last sweep — skip try: from cortexm.trace.fade import fade_sweep fade_sweep(self.store, palace=self.palace, lambda_=float(getattr(self.config, "fade_lambda", 0.05)), deactivate_threshold=float( getattr(self.config, "fade_deactivate_threshold", 0.30)), user_id=user_id) self.reader.invalidate_caches() # record the post-sweep count so the backoff anchors on what # the sweep actually left behind try: row = self.store.conn.execute( "SELECT COUNT(*) AS n FROM facts " "WHERE user_id=? AND is_active=1", (user_id,)).fetchone() post = int(row["n"]) if row else n except Exception: post = n self.store.kv_set(kv_key, str(post)) self.audit_log.log("memory.fade_pressure", resource=user_id, meta={"facts_before": n, "facts_after": post, "threshold": threshold}) except Exception as e: # never let the auto-pass kill add() self.audit_log.log("memory.fade_pressure_failed", resource=user_id, meta={"error": str(e)[:200]}) def _apply_pii(self, messages): """Run the PII guard over every message text. Returns redacted messages, or None when the policy blocks the write.""" if isinstance(messages, str): res = self.pii_guard.process(messages) if res.blocked: return None return res.redacted_text if isinstance(messages, list): out = [] for m in messages: if isinstance(m, dict) and "content" in m: res = self.pii_guard.process(str(m["content"])) if res.blocked: return None m = {**m, "content": res.redacted_text} elif isinstance(m, str): res = self.pii_guard.process(m) if res.blocked: return None m = res.redacted_text out.append(m) return out return messages def search(self, query: str, *, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None, limit: int | None = None, timestamp=None, branch: str | None = None, kind: str | None = None, **kw) -> dict: """Neuro-symbolic retrieval with full provenance. Mem0-shaped output. v0.5.3: also runs recall_step (asymmetric step-distance boost) and concatenates its context_block onto the standard search result. This is the multi_session fix — facts from scrolled-out sessions surface via the step-distance boost, not just access_count. Controlled by config.recall_step_in_search (default True). """ user_id = user_id or self.config.default_user_id ts = parse_ts(timestamp) if timestamp else None result = self.reader.search(query, user_id=user_id, agent_id=agent_id, run_id=run_id, k=limit, ts=ts, branch=branch, kind=kind) hits = result.facts and self.prefetcher.note_hits( [f.id for f in result.facts]) context_block = result.context_block # v0.5.3: wire recall_step into the production search path so all # callers benefit. The recall_step applies an asymmetric step- # distance boost to facts in danger of scrolling out of the LLM's # context window. For LongMemEval multi_session questions ("list # all the places Bob has worked"), this surfaces the OLDER # session 1 fact that the access_count boost on the current # session's fact would otherwise push below top-k. # Gate: only fire if the user has enough ingested messages for # the step-distance boost to be meaningful. Below the threshold, # recall_step would just re-rank the same top-k as search(). extra_timing = {} if getattr(self.config, "recall_step_in_search", True): try: # estimate current_step from the trace's chunk count for # this user — the most accurate proxy we have without an # explicit step counter try: n_msgs = self.store.conn.execute( "SELECT COUNT(*) FROM chunks WHERE user_id=?", (user_id,)).fetchone()[0] except Exception: n_msgs = 0 if n_msgs >= int(getattr( self.config, "recall_step_min_messages", 25)): rs = self.recall_step(query, user_id=user_id, agent_id=agent_id, run_id=run_id, current_step=n_msgs, window=int(getattr( self.config, "recall_step_window", 20)), k=int(getattr( self.config, "recall_step_k", 10))) rs_block = rs.get("context_block", "") if rs_block: context_block = (context_block + "\n\n" + rs_block if context_block else rs_block) extra_timing["recall_step"] = "ran" # union the rs results into result.facts so the # caller sees both sets in the results list rs_results = rs.get("results", []) seen_ids = {f.id for f in result.facts} for r in rs_results: # recall_step returns dicts, not Facts; pull # the underlying fact from the store if available fid = r.get("id") if fid and fid not in seen_ids: try: fobj = self.store.get_fact(fid) if fobj and fobj.is_active and not fobj.quarantined: result.facts.append(fobj) seen_ids.add(fid) except Exception: pass else: extra_timing["recall_step"] = "empty" else: extra_timing["recall_step"] = f"skipped (n_msgs={n_msgs})" except Exception as e: extra_timing["recall_step_error"] = str(e) # merge timing merged_timing = dict(result.timing) merged_timing.update(extra_timing) return { "results": result.memories(), "context_block": context_block, "relations": [{"source": f.subject, "relationship": f.relation, "destination": f.value, "valid_from": f.valid_from, "valid_to": f.valid_to} for f in result.facts], "provenance": result.provenance, "intent": result.intent, "timing": merged_timing, "llm_calls": 0, } def get_all(self, *, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None, limit: int = 200, branch: str | None = None) -> dict: user_id = user_id or self.config.default_user_id facts = self.store.query_facts(user_id=user_id, agent_id=agent_id, run_id=run_id, branch=branch, active=True, limit=limit) return {"results": [{"id": f.id, "memory": f"{f.subject} | {f.relation} | {f.value}", "event": "ADD", "valid_from": f.valid_from, "valid_to": f.valid_to, "confidence": f.confidence, "memory_type": f.memory_type, "kind": f.kind, "hash": f.source_hash} for f in facts]} def get(self, memory_id: str) -> dict | None: f = self.store.get_fact(memory_id) if not f: return None chunk = self.store.get_chunk(f.source_id) if f.source_id else None return {"id": f.id, "memory": f"{f.subject} | {f.relation} | {f.value}", "event": "ADD", "valid_from": f.valid_from, "valid_to": f.valid_to, "confidence": f.confidence, "kind": f.kind, "hash": f.source_hash, "source": chunk["text"] if chunk else None, "verified": bool(chunk and chunk["hash"] == f.source_hash)} def history(self, memory_id: str) -> list[dict]: """Full bi-temporal history of a fact chain (supersessions included).""" f = self.store.get_fact(memory_id) if not f: return [] chain = [f] seen = {f.id} frontier = [f.id] while frontier: nxt = [] for fid in frontier: for e in self.store.edges_of(fid, "CONTRADICTS", "out"): other = self.store.get_fact(e["dst"]) if other and other.id not in seen: seen.add(other.id) chain.append(other) nxt.append(other.id) for e in self.store.edges_of(fid, "CONTRADICTS", "in"): other = self.store.get_fact(e["src"]) if other and other.id not in seen: seen.add(other.id) chain.append(other) nxt.append(other.id) frontier = nxt chain.sort(key=lambda x: (x.valid_from, x.tx_from)) return [{"id": c.id, "memory": f"{c.subject} | {c.relation} | {c.value}", "event": "ADD" if c.is_active else ("QUARANTINED" if c.quarantined else "SUPERSEDED"), "valid_from": c.valid_from, "valid_to": c.valid_to, "recorded_at": c.tx_from} for c in chain] def update(self, memory_id: str, data: str, *, provenance_overlay: dict | None = None) -> dict: """Rewrite a fact's value (audit-logged, hash re-verified). ``provenance_overlay`` (mem.edit / mem.fix path): merge these keys into the fact's provenance dict on top of the standard ``manual_update`` / ``previous_value`` markers. Used by ``mem.edit()`` to stamp ``source: user_override`` so the reader can weight human-corrected facts higher than machine-extracted ones (Basic Memory learn). """ f = self.store.get_fact(memory_id) if not f: raise ContextMError(f"no fact {memory_id}") old = f.value new_prov = {**f.provenance, "manual_update": True, "previous_value": old} if provenance_overlay: new_prov = {**new_prov, **provenance_overlay} self.store.update_fact(memory_id, value=data, provenance=new_prov) self.palace.add(memory_id, self.palace.vsa.encode_fact( self.palace.embedder.embed(f.subject), self.palace.embedder.embed(f.relation), self.palace.embedder.embed(data))) return {"id": memory_id, "event": "UPDATE", "previous_value": old, "new_value": data} def delete(self, memory_id: str) -> dict: f = self.store.get_fact(memory_id) if not f: return {"id": memory_id, "event": "NOOP"} self.store.update_fact(memory_id, is_active=0, provenance={**f.provenance, "deleted": True}) return {"id": memory_id, "event": "DELETE"} def delete_all(self, *, user_id: str | None = None) -> dict: user_id = user_id or self.config.default_user_id facts = self.store.query_facts(user_id=user_id, active=True) for f in facts: self.store.update_fact(f.id, is_active=0, provenance={**f.provenance, "deleted": True}) return {"event": "DELETE_ALL", "count": len(facts)} def users(self) -> list[str]: rows = self.store.conn.execute( "SELECT DISTINCT user_id FROM facts").fetchall() return [r["user_id"] for r in rows] def reset(self) -> None: self.store.conn.executescript( "DELETE FROM facts; DELETE FROM chunks; DELETE FROM edges; " "DELETE FROM vectors; DELETE FROM commits; DELETE FROM branches; " "DELETE FROM kv;") self.store.conn.commit() self.store._ensure_genesis() self.palace = MemoryPalace(self.config, self.store) self.writer = MemoryWriter(self.config, self.store, self.palace, self.extractor) self.reader = MemoryReader(self.config, self.store, self.palace, self.prefetcher) self.git = MemoryGit(self.store, self.palace) self.zk = ZKProver(self.store, self.reader) # -------------------------------------------------- Zep temporal API def get_between(self, start, end, *, user_id: str | None = None, field: str = "valid") -> list[dict]: user_id = user_id or self.config.default_user_id s = parse_ts(start) e = parse_ts(end) facts = self.store.temporal_window( s.isoformat() if s else None, e.isoformat() if e else None, user_id=user_id, field=field, active=False) return [{"id": f.id, "fact": f.display(), "valid_from": f.valid_from, "valid_to": f.valid_to, "is_active": f.is_active} for f in facts] def get_before(self, ts, *, user_id: str | None = None, field: str = "valid") -> list[dict]: return self.get_between("1970-01-01", ts, user_id=user_id, field=field) def get_after(self, ts, *, user_id: str | None = None, field: str = "valid") -> list[dict]: return self.get_between(ts, "9999-12-31", user_id=user_id, field=field) # -------------------------------------------------- provenance & trust def audit(self, query: str, *, user_id: str | None = None) -> dict: """The 'Why' audit trail: query → VSA match → symbolic dereference → source hash → original text, for every returned fact.""" res = self.search(query, user_id=user_id) return {"query": query, "verification": res["provenance"]["verification"], "chain": res["provenance"]["chain"], "llm_calls": 0} def verify_integrity(self, sample: int | None = None) -> dict: """Recompute source hashes + vector hashes (tamper detection).""" chunks = self.store.conn.execute( "SELECT id, text, hash FROM chunks").fetchall() bad_chunks = 0 for c in chunks: if self.store.hasher.hash_text(c["text"]) != c["hash"]: bad_chunks += 1 vec = self.palace.health_check(sample) return {"chunks_checked": len(chunks), "corrupt_chunks": bad_chunks, "vector_check": {k: v for k, v in vec.items() if k != "corrupt_ids"}, "hash_provider": self.store.hasher.name, "ok": bad_chunks == 0 and vec["corrupt"] == 0} # -------------------------------------------------- Memory Git def branch(self, name: str, from_commit: str | None = None, switch: bool = True) -> str: out = self.git.branch(name, from_commit, switch) self.reader.invalidate_caches() return out def checkout(self, name: str) -> None: self.git.checkout(name) self.reader.invalidate_caches() def merge(self, name: str, strategy: str = "latest-wins") -> dict: out = self.git.merge(name, strategy) self.reader.invalidate_caches() return out def diff(self, a: str, b: str) -> dict: return self.git.diff(a, b) def blame(self, subject: str, relation: str | None = None) -> list[dict]: return self.git.blame(subject, relation) def log(self, limit: int = 50) -> list[dict]: return self.git.log(limit=limit) # -------------------------------------------------- ZK-lite def prove(self, query: str, *, user_id: str | None = None, threshold: float = 0.2) -> dict: return self.zk.prove(query, user_id=user_id or self.config.default_user_id, threshold=threshold) def verify_proof(self, proof: dict) -> bool: return self.zk.verify(proof) # -------------------------------------------------- self-healing def health_check(self, sample: int | None = None) -> dict: return self.palace.health_check(sample) def heal(self) -> dict: facts = {f.id: f for f in self.store.query_facts(active=None)} return self.palace.heal(facts) def corrupt(self, rate: float, seed: int = 0, persist: bool = False) -> int: return self.palace.corrupt(rate, seed=seed, persist=persist) # -------------------------------------------------- lifecycle & ops def apply_rules(self) -> int: """Deferred Datalog materialization after bulk ingest.""" self.store.begin_batch() derived = self.writer.apply_rules() self.reader.invalidate_caches() return len(derived) # -------------------------------------------------- async enrichment def enrich(self, user_id: str | None = None, *, extractor=None, limit: int | None = None, min_confidence: float | None = None, dry_run: bool = False) -> dict: """Post-store LLM enrichment fallback (μ=0 stays intact on the synchronous ingest path; this is the plan's graceful-degradation second pass over zero-signal chunks). Returns an EnrichmentReport as a dict — LLM call count and provenance markers are auditable. """ from cortexm.bridge.enrich import enrich as _enrich rep = _enrich(self.writer, user_id, extractor=extractor, limit=limit, min_confidence=min_confidence, dry_run=dry_run) self.reader.invalidate_caches() return rep.to_dict() def enrich_async(self, user_id: str | None = None, **kw): """Background-thread variant of enrich(). Returns (thread, holder).""" from cortexm.bridge.enrich import enrich_async as _ea return _ea(self.writer, user_id, **kw) # -------------------------------------------------- scope sandbox def promote(self, fact_ids: list[str], *, reviewed_by: str = "system", force: bool = False) -> dict: """Promote agent-scoped facts into the user scope (InjecMEM policy). Gated on confidence + a fresh InjecMEM/MINJA rescan of the source chunk; every decision lands in the tamper-evident audit chain. """ from cortexm.security.sandbox import ScopeSandbox sandbox = ScopeSandbox(self.config, self.store, self.audit_log) out = sandbox.promote(fact_ids, reviewed_by=reviewed_by, force=force) self.reader.invalidate_caches() return out def consolidate(self, now=None, **kwargs) -> dict: """Run BOTH consolidation passes: (1) lifecycle.consolidate — Dual-Layer Agentic Memory: promote reinforced short-term facts, decay untouched ones, demote weak long-term facts. Fast, pure-SQL. (2) trace.consolidate.consolidate — Aeon-inspired "dreaming": merge redundant triples (MERGED_WITH edges), retire stale facts past valid_to + grace, defrag palace, retrain MBTB prefetcher. Slower, idempotent, safe. (3) trace.cognition — HMS-style self-organization: PatternScanner + AbstractionEngine + GapDetector + HypothesisEngine + AnalogyDetector. Emits HYPOTHESIZED_BY edges with confidence < 0.5 — never promoted to active retrieval unless explicitly confirmed by user input. Returns a combined report. Either pass may be skipped via kwargs lifecycle=False / dreaming=False / cognition=False. """ out = {"lifecycle": {}, "dreaming": {}} if kwargs.get("lifecycle", True): out["lifecycle"] = lifecycle.consolidate(self.store, now) if kwargs.get("dreaming", True): from cortexm.trace.consolidate import consolidate as _dream # Respect the Config's fade_enabled / tmt_enabled / # cognition_enabled flags by default so `cortexm # consolidate` runs the full production pass without # requiring CLI flag plumbing. CLI / env can still turn # them off via kwargs. run_fade = kwargs.get("run_fade", getattr(self.config, "fade_enabled", True)) run_tmt = kwargs.get("run_tmt", getattr(self.config, "tmt_enabled", False)) run_cognition = kwargs.get( "run_cognition", getattr(self.config, "cognition_enabled", False)) out["dreaming"] = _dream(self.store, palace=self.palace, prefetcher=self.prefetcher, user_id=kwargs.get("user_id"), dry_run=kwargs.get("dry_run", False), run_fade=run_fade, run_tmt=run_tmt, run_cognition=run_cognition) # v0.6.8: wiki-as-truth — human-ownable markdown export on # every consolidate when wiki_dir is configured. Best-effort: # never break consolidation on an export failure. wiki_dir = getattr(self.config, "wiki_dir", None) if wiki_dir and not kwargs.get("dry_run", False): try: rep = self.export_markdown( wiki_dir, user_id=kwargs.get("user_id")) out["wiki_export"] = rep except Exception as e: # noqa: BLE001 — export never breaks consolidate out["wiki_export"] = {"error": str(e)[:200]} return out def export_schema_report(self, user_id: str | None = None) -> dict: return export_schema_report(self.store, user_id) @staticmethod def merge_schema_reports(reports: list[dict]) -> dict: return merge_schema_reports(reports) # -------------------------------------------------- introspection def stats(self) -> dict: s = self.store.stats() s.update(self.palace.storage_stats()) s.update(self.reader.slb.stats()) s.update(self.prefetcher.stats()) s["counters"] = metrics.counters() s["u0_protocol"] = "verified" if metrics.llm_calls() == 0 else "VIOLATED" s["hash_provider"] = self.store.hasher.name s["vsa_mode"] = self.config.vsa_mode s["dims"] = self.config.dims s["pii_mode"] = self.config.pii_mode return s def storage_stats(self) -> dict: return self.palace.storage_stats() # -------------------------------------------------- sidecar blob arena def enable_blob_arena(self, path: str | os.PathLike) -> dict: """Opt-in: migrate chunks.text into a sidecar mmap-backed blob file (Aeon-inspired). After migration: - chunks.text := first 64 bytes of the source (preview) - chunks.blob_offset := byte offset in the arena file - chunks.blob_len := payload length - chunks.blob_compressed := 0/1 Graph queries that only need the preview still work without touching the arena. Full text is fetched on demand via arena.get_text(offset, len, compressed). Returns a migration report dict. """ from cortexm.trace.blob_arena import ( BlobArena, migrate_chunks_to_arena) arena = BlobArena(path) report = migrate_chunks_to_arena(self.store, arena) # keep arena handle on self so it stays alive for the life of # the Memory instance; the host can grab it via .blob_arena self.blob_arena = arena return report def get_chunk_text(self, chunk_id: str) -> str: """Fetch full text for a chunk — from the arena if migrated, otherwise from the inline text column.""" arena = getattr(self, "blob_arena", None) if arena is not None: from cortexm.trace.blob_arena import get_chunk_text as _g return _g(self.store, arena, chunk_id) row = self.store.conn.execute( "SELECT text FROM chunks WHERE id=?", (chunk_id,)).fetchone() return row[0] if row else "" # ------------------------------------- engineered role vectors (NSR) def use_engineered_role_vectors(self, *, n_epochs: int = 200, lr: float = 0.01, save_path: str | None = None, verbose: bool = False) -> dict: """NSR-inspired: train a tiny autoencoder on the actual fact corpus and use the top-k principal directions as role vectors. arXiv insight: random role vectors (current default) waste capacity on directions orthogonal to the data. Engineered ones sit on the data's principal axes — higher effective capacity, lower cross-talk, better retrieval SNR. After this call, the palace's VSA will use the engineered role vectors for bind() / unbind() / probe() operations. New facts ingested after this call will be encoded with the engineered vectors; previously-encoded facts keep their original (random) holograms until the palace is rebuilt. Returns the autoencoder training report. NOTE: this is OPT-IN. The default behavior (random role vectors) is unchanged unless this method is called. """ from cortexm.vsa.role_vectors import EngineeredRoleVectors # pull the fact matrix from the palace — we need the actual # S/R/V vectors used to encode the existing facts facts = self.store.query_facts(active=True) if not facts: return {"trained": False, "reason": "no_facts_in_store"} # build a (n_facts * 3, dims) matrix: S, R, V vectors for each # fact, stacked. Each row is a single role-filler vector. rows = [] for f in facts: for field, role in ( (f.subject, "S"), (f.relation, "R"), (f.value, "V")): if field: rows.append(self.palace.embedder.embed(field)) if not rows: return {"trained": False, "reason": "no_text_to_embed"} matrix = np.stack(rows).astype(np.float32) # cap the matrix size — 10k samples is plenty for PCA if len(matrix) > 10_000: import numpy as _np rng = _np.random.default_rng(self.config.seed) idx = rng.choice(len(matrix), 10_000, replace=False) matrix = matrix[idx] erv = EngineeredRoleVectors( dims=self.config.dims, n_roles=3, seed=self.config.seed, n_epochs=n_epochs, lr=lr) report = erv.fit(matrix) if erv.is_fit: self.palace.vsa.use_engineered(erv) self._engineered_role_vectors = erv if save_path: erv.save(save_path) if verbose: print(f"[engineered-role-vectors] {report}") return report def save_engineered_role_vectors(self, path: str) -> None: erv = getattr(self, "_engineered_role_vectors", None) if erv is None: raise RuntimeError("no engineered role vectors to save — " "call use_engineered_role_vectors() first") erv.save(path) def load_engineered_role_vectors(self, path: str) -> dict: """Load a previously-saved .npz of engineered role vectors and swap them in as the active role vectors.""" from cortexm.vsa.role_vectors import EngineeredRoleVectors erv = EngineeredRoleVectors(dims=self.config.dims, seed=self.config.seed) erv.load(path) if erv.is_fit: self.palace.vsa.use_engineered(erv) self._engineered_role_vectors = erv return {"loaded": True, "path": path} return {"loaded": False, "reason": "file_empty_or_corrupt"} def close(self) -> None: # Close sidecar arena first on Windows (mmap holds file lock, prevents unlink) try: arena = getattr(self, "blob_arena", None) if arena is not None: arena.close() except Exception: pass self.palace.close() self.store.close() # ------------------------------------------------------------------ # v0.6.1: BM25 + index maintenance facade # Exposed on Memory so users can call ``m.tune_bm25(k1=1.2, b=0.5)`` # and ``m.optimize_index()`` without reaching into the verbatim # plugin. If the verbatim tier isn't mounted (e.g. disabled in # Config), these are no-ops so callers don't need to defensively # check. def tune_bm25(self, k1: float = 1.2, b: float = 0.75) -> None: """Tune BM25 k1 (term saturation) + b (length normalization). Lucene defaults: k1=1.2, b=0.75. Our corpus (short chat chunks, avg ~12 tokens) tends to prefer slightly higher saturation and weaker length norm; the v0.6.0 defaults were k1=1.5, b=0.75. Run ``scripts/tune_bm25_canonical.py`` to grid-search on the canonical LongMemEval sample and pick the best for your data. Takes effect on the next ``search()`` call. """ if self._verbatim is not None: self._verbatim.tune_bm25(k1=k1, b=b) # Persist on the config too so reopens honor the tuning. self.config.bm25_k1 = k1 self.config.bm25_b = b def optimize_index(self) -> dict: """VACUUM + FTS5 optimize + WAL checkpoint. Run this after large bulk ingests to fold the WAL back into the main .db file, reclaim deleted-page space, and merge the FTS5 b-tree segments. Idempotent; safe to call any time. """ if self._verbatim is not None: return self._verbatim.optimize_index() return {"optimized": False, "reason": "verbatim tier not mounted"} def _reopen(self) -> None: """Rebind every component to a freshly-opened store (post-restore).""" from cortexm.security.pii import PIIGuard, PIIVault from cortexm.security.rbac import APIKeyStore from cortexm.enterprise.audit import AuditLog from cortexm.enterprise.governance import Governance self.store = TraceStore(self.config.db_path, HashProvider(self.config.hash_provider), wal_sync=getattr(self.config, "wal_sync", "normal"), pragma_cache_mb=getattr(self.config, "pragma_cache_mb", 64), pragma_mmap_mb=getattr(self.config, "pragma_mmap_mb", 256), pragma_threads=getattr(self.config, "pragma_threads", 4), pragma_temp_in_memory=getattr(self.config, "pragma_temp_in_memory", True), pragma_locking_exclusive=getattr(self.config, "pragma_locking_exclusive", False)) self.palace = MemoryPalace(self.config, self.store) self.writer = MemoryWriter(self.config, self.store, self.palace, self.extractor) self.reader = MemoryReader(self.config, self.store, self.palace, self.prefetcher) self.git = MemoryGit(self.store, self.palace) self.zk = ZKProver(self.store, self.reader) self.pii_vault = PIIVault(self.store, self.cipher) self.pii_guard = PIIGuard(self.config.pii_mode, self.pii_vault) self.keys = APIKeyStore(self.store) self.audit_log = AuditLog(self.store, enabled=self.config.audit_enabled) self.governance = Governance(self) # the governance object captured the OLD store/palace in __init__; # rebind to the fresh ones self.governance.store = self.store self.governance.palace = self.palace self.governance.audit = self.audit_log def __enter__(self) -> "Memory": return self def __exit__(self, *exc) -> None: self.close()