""" Title: Total Recall (Persistent JSON Memory Engine) Author: BobbyLLM Version: 1.0.0 Description: High-performance minimal persistent memory system for OpenWebUI. Features include: • Chronological recall • Optional timestamp recall • Second-person reframing • Update memory by index • Delete memory by index • Move memory entries into a dedicated "done" archive • Restore entries from "done" back into active memory • Full JSON export/import (raw facts.json) • Auto-fix JSON imports with safe atomic writes • Strong LLM-guidance docstrings to prevent hallucinated updates/deletes """ from __future__ import annotations import os, re, time, datetime as _dt from typing import Dict, Any from pydantic import BaseModel, Field # ------------------------------------------------------------ # JSON Backend (fast or fallback) # ------------------------------------------------------------ try: import orjson as _json def _dumps(obj): return _json.dumps(obj).decode("utf-8") def _loads(s): return _json.loads(s) except Exception: import json as _json def _dumps(obj): return _json.dumps(obj, ensure_ascii=False) def _loads(s): return _json.loads(s) # ------------------------------------------------------------ # Timestamp helper # ------------------------------------------------------------ def _ts(): return _dt.datetime.now().astimezone().strftime("%B %d, %Y at %H:%M:%S %Z") # ------------------------------------------------------------ # Utility helpers # ------------------------------------------------------------ def _ensure_dir(path: str): os.makedirs(path, exist_ok=True) return path def _safe_key(s: str): s = s.strip().lower() s = re.sub(r"[^a-z0-9_]+", "_", s) return s or f"key_{int(time.time())}" def _reframe_second_person(text: str) -> str: rep = [ (" I'm ", " you're "), (" I am ", " you are "), (" I was ", " you were "), (" I ", " you "), (" my ", " your "), (" mine ", " yours "), (" me ", " you "), (" myself ", " yourself "), ] t = f" {text} " for old, new in rep: t = t.replace(old, new) return t.strip() def extract_facts(text: str) -> Dict[str, str]: patterns = [ r"(?i)my ([a-zA-Z0-9_ ]+) is ([^\.]+)", r"(?i)([a-zA-Z0-9_ ]+) is ([^\.]+)", r"(?i)I have ([a-zA-Z0-9_ ]+) ([^\.]+)", ] out = {} for pat in patterns: for m in re.findall(pat, text): if isinstance(m, tuple) and len(m) == 2: k, v = m out[_safe_key(k)] = v.strip() return out # ------------------------------------------------------------ # Storage Engine # ------------------------------------------------------------ class Storage: def __init__(self, base_dir: str, debug: bool = False): self.base = _ensure_dir(os.path.join(base_dir, "total_recall")) self.facts_file = os.path.join(self.base, "facts.json") self.ids_file = os.path.join(self.base, "ids.json") self.done_dir = _ensure_dir(os.path.join(self.base, "done")) self.debug_log = os.path.join(self.base, "debug.log") self.debug_enabled = debug self._facts_cache = None self._ids_cache = None if not os.path.exists(self.facts_file): self._atomic_write(self.facts_file, {}) if not os.path.exists(self.ids_file): self._atomic_write(self.ids_file, {"last_id": 0}) def debug(self, *args): if not self.debug_enabled: return msg = " ".join(str(a) for a in args) with open(self.debug_log, "a", encoding="utf-8") as f: f.write(f"{_ts()} {msg}\n") def _atomic_write(self, path: str, obj: Any): tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: f.write(_dumps(obj)) os.replace(tmp, path) def load_facts(self): if self._facts_cache is None: with open(self.facts_file, "r", encoding="utf-8") as f: self._facts_cache = _loads(f.read()) return self._facts_cache def save_facts(self, data): self._facts_cache = data self._atomic_write(self.facts_file, data) def load_ids(self): if self._ids_cache is None: with open(self.ids_file, "r", encoding="utf-8") as f: self._ids_cache = _loads(f.read()) return self._ids_cache def save_ids(self, data): self._ids_cache = data self._atomic_write(self.ids_file, data) def next_id(self) -> int: ids = self.load_ids() nid = ids.get("last_id", 0) + 1 ids["last_id"] = nid self.save_ids(ids) return nid def write_done(self, rid: int, row: Dict[str, Any]): now = _dt.datetime.now().astimezone() folder = _ensure_dir( os.path.join(self.done_dir, f"{now.year}", f"{now.month:02d}") ) path = os.path.join(folder, f"{rid:08d}.json") self._atomic_write(path, row) def list_done(self): out = [] for root, _, files in os.walk(self.done_dir): for f in files: if f.endswith(".json"): try: with open(os.path.join(root, f), "r", encoding="utf-8") as fh: out.append(_loads(fh.read())) except: pass out.sort(key=lambda x: x.get("created_at", ""), reverse=True) return out def delete_done(self, rid: int): filename = f"{rid:08d}.json" for root, _, files in os.walk(self.done_dir): if filename in files: os.remove(os.path.join(root, filename)) break # ------------------------------------------------------------ # Tools Interface # ------------------------------------------------------------ class Tools: id = "total_recall" name = "Total Recall" author = "BobbyLLM" version = "1.0.2" class Valves(BaseModel): storage_dir: str = Field( default="", description="Directory for Total Recall storage." ) debug: bool = Field(default=False, description="Enable debug logging.") def __init__(self): self.valves = self.Valves() def _store(self) -> Storage: base = self.valves.storage_dir or os.getenv("DATA_DIR") or os.getcwd() return Storage(base, debug=self.valves.debug) # -------------------------------------------------------- # ADD FACTS # -------------------------------------------------------- def add_facts(self, text: str, __user__=None, __metadata__=None): """ Store a new memory value. LLM Guidance: • You MUST call this tool when the user says: "remember this", "store this", "add this to memory", etc. • Do NOT claim memory was stored unless this tool executes. """ S = self._store() extracted = extract_facts(text) if not extracted: extracted = {_safe_key(f"fact_{int(time.time())}"): text} data = S.load_facts() for k, v in extracted.items(): data[k] = {"value": v, "updated_at": _ts()} S.save_facts(data) return {"ok": True, "stored": extracted} # -------------------------------------------------------- # LIST FACTS (NO TIMESTAMPS) # -------------------------------------------------------- def list_facts(self, __user__=None): """ List memories WITHOUT timestamps. LLM Guidance: • Use ONLY when user does NOT ask for dates/times. """ S = self._store() data = S.load_facts() items = sorted(data.items(), key=lambda kv: kv[1]["updated_at"]) return { "facts": [_reframe_second_person(rec["value"]) for _, rec in items], "count": len(items), } # -------------------------------------------------------- # LIST FACTS WITH TIMESTAMPS # -------------------------------------------------------- def list_facts_with_timestamps(self, __user__=None): """ List memories WITH timestamps. LLM Guidance: • MUST be used when the user asks: "when", "what time", "timestamp", "chronological", etc. """ S = self._store() data = S.load_facts() items = sorted(data.items(), key=lambda kv: kv[1]["updated_at"]) return { "facts": [ { "value": _reframe_second_person(rec["value"]), "updated_at": rec["updated_at"], } for _, rec in items ], "count": len(items), } # -------------------------------------------------------- # DELETE FACT BY INDEX # -------------------------------------------------------- def delete_fact_index(self, index: int, __user__=None): """ Delete memory by chronological index. LLM Guidance: • MUST be called to delete memory. • Do NOT hallucinate deletion. • Return the tool’s JSON so the user can verify actual removal. """ S = self._store() data = S.load_facts() items = sorted(data.items(), key=lambda kv: kv[1]["updated_at"]) if index < 1 or index > len(items): return {"ok": False, "deleted": False, "error": "Index out of range"} key, _ = items[index - 1] del data[key] S.save_facts(data) return {"ok": True, "deleted": True, "index": index, "key": key} # -------------------------------------------------------- # UPDATE FACT BY INDEX # -------------------------------------------------------- def update_fact_index(self, index: int, new_value: str, __user__=None): """ Update the VALUE of memory by index. LLM Guidance: • MUST be called when user requests edits/corrections. """ S = self._store() data = S.load_facts() items = sorted(data.items(), key=lambda kv: kv[1]["updated_at"]) if index < 1 or index > len(items): return {"ok": False, "updated": False, "error": "Index out of range"} key, _ = items[index - 1] data[key] = {"value": new_value, "updated_at": _ts()} S.save_facts(data) return {"ok": True, "updated": True, "index": index, "key": key} # -------------------------------------------------------- # MOVE FACT TO DONE # -------------------------------------------------------- def move_fact_to_done(self, index: int, __user__=None): """ Move a memory entry into the done/archive folder. LLM Guidance: • MUST be called when the user says: "mark #X done", "complete index X", "move X to done". • Do NOT claim completion unless this tool executes. """ S = self._store() data = S.load_facts() items = sorted(data.items(), key=lambda kv: kv[1]["updated_at"]) if index < 1 or index > len(items): return {"ok": False, "moved": False, "error": "Index out of range"} key, rec = items[index - 1] value = rec["value"] # delete active entry del data[key] S.save_facts(data) # write to done folder rid = S.next_id() row = { "id": rid, "created_at": _ts(), "value": value, "original_key": key, "status": "done", } S.write_done(rid, row) return { "ok": True, "moved": True, "index": index, "done_id": rid, "value": value, } # -------------------------------------------------------- # RESTORE DONE ENTRY BACK TO MEMORY # -------------------------------------------------------- def restore_done_index(self, index: int, __user__=None): """ Restore a done/archive entry back into active memory. LLM Guidance: • MUST be used when the user says: "bring #X back", "restore X", "undo done X", etc. • Do NOT claim restoration unless this tool executes. """ S = self._store() done_list = S.list_done() if index < 1 or index > len(done_list): return {"ok": False, "restored": False, "error": "Index out of range"} entry = done_list[index - 1] rid = entry.get("id") value = entry.get("value", "") # delete from done folder S.delete_done(rid) # re-add to active memory data = S.load_facts() new_key = _safe_key(f"restored_{rid}") data[new_key] = {"value": value, "updated_at": _ts()} S.save_facts(data) return { "ok": True, "restored": True, "done_index": index, "restored_key": new_key, "value": value, } # -------------------------------------------------------- # EXPORT RAW JSON # -------------------------------------------------------- def export_facts(self, __user__=None): """ Export raw internal facts.json for manual editing. LLM Guidance: • MUST be used when user asks to edit memory JSON directly. """ S = self._store() return {"json": S.load_facts()} # -------------------------------------------------------- # IMPORT RAW JSON WITH AUTO-FIX # -------------------------------------------------------- def import_facts(self, json_string: str, __user__=None): """ Import edited JSON safely with auto-fix. Auto-fix: • normalizes bad keys • ensures values and timestamps exist • prevents malformed data from breaking storage • atomic write safety LLM Guidance: • Use ONLY when user explicitly supplies edited JSON. """ S = self._store() try: new_data = _loads(json_string) except Exception as e: return {"ok": False, "error": f"Invalid JSON: {e}"} fixed = {} for k, v in new_data.items(): key = _safe_key(k) if isinstance(v, dict) and "value" in v: val = str(v["value"]) ts = v.get("updated_at", _ts()) else: val = str(v) ts = _ts() fixed[key] = {"value": val, "updated_at": ts} S.save_facts(fixed) return {"ok": True, "imported": True, "count": len(fixed)} # -------------------------------------------------------- # DONE LIST / GET / DELETE # -------------------------------------------------------- def list_done(self, __user__=None): S = self._store() return S.list_done() def get_done(self, rid: int, __user__=None): S = self._store() for item in S.list_done(): if int(item.get("id", -1)) == rid: return item return {} def delete_done(self, rid: int, __user__=None): S = self._store() S.delete_done(rid) return {"ok": True}