""" VERITAS Integrated Memory System ================================== CogniVault (RAG) + Aluna Memory (LTM) — unified HuggingFace Space Rob "The Sounds Guy" Barenbrug | Built by VERITAS Stack: Python 3.11 + Gradio 5.15.0 """ import os, json, sqlite3, hashlib, datetime, re, logging from pathlib import Path import gradio as gr logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("veritas") # ─── STORAGE ───────────────────────────────────────────────────────────────── DB_PATH = Path("/data/veritas.db") if Path("/data").exists() else Path("veritas.db") DATA_DIR = Path("/data/uploads") if Path("/data").exists() else Path("uploads") DATA_DIR.mkdir(parents=True, exist_ok=True) def init_db(): with sqlite3.connect(str(DB_PATH)) as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT NOT NULL, category TEXT DEFAULT 'general', importance INTEGER DEFAULT 5, source TEXT DEFAULT 'manual', created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')), access_count INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS knowledge ( id TEXT PRIMARY KEY, title TEXT, content TEXT NOT NULL, source TEXT DEFAULT 'manual', tags TEXT DEFAULT '[]', doc_type TEXT DEFAULT 'text', chunk_index INTEGER DEFAULT 0, parent_id TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE INDEX IF NOT EXISTS idx_mem_cat ON memories(category); CREATE INDEX IF NOT EXISTS idx_mem_imp ON memories(importance DESC); CREATE INDEX IF NOT EXISTS idx_know_src ON knowledge(source); """) init_db() def make_id(s: str) -> str: return hashlib.sha256((s + datetime.datetime.utcnow().isoformat()).encode()).hexdigest()[:20] # ─── MEMORY OPS ────────────────────────────────────────────────────────────── def store_memory(content, category="general", importance=5, source="manual"): if not str(content).strip(): return {"error": "Content cannot be empty"} with sqlite3.connect(str(DB_PATH)) as conn: row = conn.execute("SELECT id FROM memories WHERE content=?", (content,)).fetchone() if row: conn.execute("UPDATE memories SET access_count=access_count+1, updated_at=datetime('now') WHERE id=?", (row[0],)) return {"id": row[0], "status": "deduplicated", "message": "Already stored — access count bumped"} mid = make_id(content) conn.execute("INSERT INTO memories (id, content, category, importance, source) VALUES (?,?,?,?,?)", (mid, content, category, int(importance), source)) return {"id": mid, "status": "stored", "category": category, "importance": importance} def search_memories(query, category=None, limit=20): q = f"%{query.lower()}%" with sqlite3.connect(str(DB_PATH)) as conn: if category and category != "all": rows = conn.execute( "SELECT id,content,category,importance,created_at,access_count FROM memories " "WHERE lower(content) LIKE ? AND category=? ORDER BY importance DESC,access_count DESC LIMIT ?", (q, category, limit)).fetchall() else: rows = conn.execute( "SELECT id,content,category,importance,created_at,access_count FROM memories " "WHERE lower(content) LIKE ? ORDER BY importance DESC,access_count DESC LIMIT ?", (q, limit)).fetchall() ids = [r[0] for r in rows] if ids: conn.execute(f"UPDATE memories SET access_count=access_count+1 WHERE id IN ({','.join('?'*len(ids))})", ids) return [{"id":r[0],"content":r[1],"category":r[2],"importance":r[3],"created_at":r[4],"access_count":r[5]} for r in rows] def get_all_memories(limit=100): with sqlite3.connect(str(DB_PATH)) as conn: rows = conn.execute( "SELECT id,content,category,importance,created_at,access_count FROM memories " "ORDER BY importance DESC,updated_at DESC LIMIT ?", (limit,)).fetchall() return [{"id":r[0],"content":r[1],"category":r[2],"importance":r[3],"created_at":r[4],"access_count":r[5]} for r in rows] # ─── KNOWLEDGE OPS ─────────────────────────────────────────────────────────── def chunk_text(text, size=800, overlap=100): paras = [p.strip() for p in re.split(r'\n\n+', text) if p.strip()] chunks, cur = [], "" for p in paras: if len(cur) + len(p) + 2 < size: cur = (cur + "\n\n" + p).strip() if cur else p else: if cur: chunks.append(cur) cur = p if cur: chunks.append(cur) final = [] for c in chunks: if len(c) > size * 1.5: for i in range(0, len(c), size - overlap): piece = c[i:i+size] if piece.strip(): final.append(piece) else: final.append(c) return final or [text] def ingest_text(title, content, source="manual", tags_str="", doc_type="text"): if not str(content).strip(): return {"error": "Content is empty"} tags = [t.strip() for t in str(tags_str).split(",") if t.strip()] chunks = chunk_text(str(content)) parent_id = make_id(str(title) + str(content)[:80]) with sqlite3.connect(str(DB_PATH)) as conn: for i, chunk in enumerate(chunks): kid = make_id(chunk + str(i)) conn.execute( "INSERT OR REPLACE INTO knowledge (id,title,content,source,tags,doc_type,chunk_index,parent_id) VALUES (?,?,?,?,?,?,?,?)", (kid, str(title), chunk, source, json.dumps(tags), doc_type, i, parent_id)) return {"parent_id": parent_id, "title": str(title), "chunks": len(chunks), "chars": len(str(content)), "source": source, "tags": tags, "status": "ingested"} def ingest_file(file_obj, source="upload"): if file_obj is None: return {"error": "No file provided"} # Gradio 5 returns filepath string or dict if isinstance(file_obj, dict): filepath = file_obj.get("name", file_obj.get("path", "")) filename = file_obj.get("orig_name", Path(filepath).name) else: filepath = str(file_obj) filename = Path(filepath).name ext = Path(filename).suffix.lower() if ext not in [".txt", ".md"]: return {"error": f"Unsupported: {ext}. Use .txt or .md"} try: with open(filepath, "r", encoding="utf-8", errors="replace") as f: content = f.read() return ingest_text(filename, content, source=source, tags_str=f"{ext.strip('.')},{source}", doc_type=ext.strip(".")) except Exception as e: return {"error": str(e)} def search_knowledge(query, limit=20): q = f"%{query.lower()}%" with sqlite3.connect(str(DB_PATH)) as conn: rows = conn.execute( "SELECT id,title,content,source,tags,doc_type,created_at FROM knowledge " "WHERE lower(content) LIKE ? OR lower(title) LIKE ? ORDER BY created_at DESC LIMIT ?", (q, q, limit)).fetchall() return [{"id":r[0],"title":r[1],"content":r[2],"source":r[3],"tags":json.loads(r[4]),"doc_type":r[5],"created_at":r[6]} for r in rows] def rag_query(question): if not str(question).strip(): return "Enter a question above." know = search_knowledge(question, 6) mems = search_memories(question, limit=4) if not know and not mems: return "❌ No context found. Add documents via 📚 Knowledge or memories via 🧠 Memory tab." parts = [] if know: parts.append("## Knowledge Base\n") for r in know[:5]: parts.append(f"**[{r['title']}]** (source: {r['source']})\n{r['content'][:500]}\n---") if mems: parts.append("\n## Memory Context\n") for m in mems[:3]: parts.append(f"**[{m['category']} | importance {m['importance']}]**\n{m['content'][:300]}\n---") return "\n".join(parts) def get_stats(): with sqlite3.connect(str(DB_PATH)) as conn: mc = conn.execute("SELECT COUNT(*) FROM memories").fetchone()[0] kc = conn.execute("SELECT COUNT(*) FROM knowledge").fetchone()[0] cats = dict(conn.execute("SELECT category,COUNT(*) FROM memories GROUP BY category").fetchall()) srcs = dict(conn.execute("SELECT source,COUNT(*) FROM knowledge GROUP BY source").fetchall()) top = conn.execute("SELECT content,importance,access_count FROM memories ORDER BY importance DESC,access_count DESC LIMIT 5").fetchall() return {"total_memories": mc, "total_knowledge_chunks": kc, "categories": cats, "knowledge_sources": srcs, "top_memories": [{"content": r[0][:120], "importance": r[1], "access_count": r[2]} for r in top]} # ─── MCP JSON API ───────────────────────────────────────────────────────────── def mcp_api(request_json): try: req = json.loads(request_json) except: return json.dumps({"error": "Invalid JSON"}, indent=2) tool = req.get("tool", "") params = req.get("params", {}) if tool == "store_memory": result = store_memory(params.get("content",""), params.get("category","general"), int(params.get("importance",5)), params.get("source","claude")) elif tool == "search_memories": result = search_memories(params.get("query",""), params.get("category"), int(params.get("limit",20))) elif tool == "store_knowledge": result = ingest_text(params.get("title","Untitled"), params.get("content",""), params.get("source","claude"), params.get("tags",""), params.get("doc_type","text")) elif tool == "search_knowledge": result = search_knowledge(params.get("query",""), int(params.get("limit",20))) elif tool == "rag_query": result = {"context": rag_query(params.get("question",""))} elif tool == "get_session_context": result = {"user":"Rob 'The Sounds Guy' Barenbrug","location":"Durban, South Africa", "device":"Huawei Pura 80 Pro (Termux)","vps":"veritas.alunaafrica.cloud", "constraint":"CANNOT manually code — click-and-run only", "philosophy":"Live in truth, never in comfort", "stats": get_stats(), "recent_memories": get_all_memories(10)} elif tool == "status": result = get_stats() result.update({"system":"VERITAS Integrated Memory","version":"1.0.0","status":"OPERATIONAL","stack":"Python 3.11 + Gradio 5.15.0"}) else: result = {"error": f"Unknown tool: '{tool}'", "available": ["store_memory","search_memories","store_knowledge","search_knowledge","rag_query","get_session_context","status"]} return json.dumps(result, indent=2, default=str) # ─── UI ─────────────────────────────────────────────────────────────────────── CATS = ["general","project","technical","personal","rib-rage","bytebot","cognivault","aluna","android","audio","business","mene-portal"] with gr.Blocks(title="VERITAS Memory", theme=gr.themes.Base(primary_hue="blue", neutral_hue="slate"), css="footer{display:none!important}.header{text-align:center;padding:16px 0 6px}") as demo: gr.HTML("""

🔷 VERITAS Memory System

CogniVault RAG + Aluna Memory LTM — Unified

Rob "The Sounds Guy" Barenbrug  |  Live in truth, never in comfort

""") with gr.Tabs(): # TAB 1 — MEMORY with gr.Tab("🧠 Memory"): gr.Markdown("### Store long-term memories") with gr.Row(): with gr.Column(scale=3): t_content = gr.Textbox(label="Memory Content", placeholder="What should be remembered…", lines=3) with gr.Column(scale=1): t_cat = gr.Dropdown(CATS, value="general", label="Category") t_imp = gr.Slider(1, 10, value=5, step=1, label="Importance") with gr.Row(): t_store_btn = gr.Button("💾 Store Memory", variant="primary") t_clear_btn = gr.Button("Clear") t_store_out = gr.JSON(label="Result") gr.Markdown("---\n### Search Memories") with gr.Row(): t_sq = gr.Textbox(label="Search", placeholder="Search memories…", scale=3) t_scat = gr.Dropdown(["all"] + CATS, value="all", label="Filter", scale=1) t_sbtn = gr.Button("🔍 Search", variant="primary", scale=1) t_sout = gr.Dataframe(headers=["content","category","importance","access_count","created_at"], label="Results", wrap=True) def fn_store(c, cat, imp): return store_memory(c, cat, imp), "" def fn_search_m(q, cat): r = search_memories(q, None if cat=="all" else cat) return [[x["content"][:200],x["category"],x["importance"],x["access_count"],x["created_at"]] for x in r] if r else [] t_store_btn.click(fn_store, [t_content, t_cat, t_imp], [t_store_out, t_content]) t_clear_btn.click(lambda: ("", None), outputs=[t_content, t_store_out]) t_sbtn.click(fn_search_m, [t_sq, t_scat], t_sout) # TAB 2 — KNOWLEDGE with gr.Tab("📚 Knowledge"): gr.Markdown("### Ingest documents into CogniVault RAG") with gr.Tabs(): with gr.Tab("📝 Paste Text"): k_title = gr.Textbox(label="Title", placeholder="Bytebot Architecture Notes") k_content = gr.Textbox(label="Content", lines=7, placeholder="Paste text, notes, WhatsApp exports…") k_source = gr.Textbox(label="Source", value="manual") k_tags = gr.Textbox(label="Tags (comma-separated)", placeholder="bytebot, deployment") k_btn = gr.Button("📥 Ingest", variant="primary") k_out = gr.JSON(label="Result") k_btn.click(lambda ti,co,so,ta: ingest_text(ti,co,so,ta), [k_title,k_content,k_source,k_tags], k_out) with gr.Tab("📁 Upload File"): f_file = gr.File(label="Upload .txt or .md", file_types=[".txt",".md"]) f_source = gr.Textbox(label="Source", value="upload") f_btn = gr.Button("📥 Ingest File", variant="primary") f_out = gr.JSON(label="Result") f_btn.click(ingest_file, [f_file, f_source], f_out) gr.Markdown("---\n### Search Knowledge") with gr.Row(): ks_q = gr.Textbox(label="Search Knowledge", placeholder="Search docs…", scale=3) ks_btn = gr.Button("🔍 Search", variant="primary", scale=1) ks_out = gr.Dataframe(headers=["title","content","source","doc_type","created_at"], label="Knowledge Results", wrap=True) def fn_sk(q): r = search_knowledge(q) return [[x["title"],x["content"][:250],x["source"],x["doc_type"],x["created_at"]] for x in r] if r else [] ks_btn.click(fn_sk, ks_q, ks_out) # TAB 3 — RAG with gr.Tab("🔮 RAG Query"): gr.Markdown("### Ask a question — get context to paste into Claude") r_q = gr.Textbox(label="Question", placeholder="What is the Bytebot deployment process?", lines=2) r_btn = gr.Button("🔮 Retrieve Context", variant="primary") r_out = gr.Textbox(label="Retrieved Context — copy → paste into Claude", lines=16, show_copy_button=True) r_btn.click(rag_query, r_q, r_out) # TAB 4 — DASHBOARD with gr.Tab("📊 Dashboard"): d_btn = gr.Button("🔄 Refresh", variant="secondary") with gr.Row(): d_mc = gr.Number(label="Total Memories", interactive=False) d_kc = gr.Number(label="Knowledge Chunks", interactive=False) d_top = gr.Dataframe(headers=["content","importance","access_count"], label="Top Memories", wrap=True) d_cats = gr.JSON(label="Memory Categories") d_src = gr.JSON(label="Knowledge Sources") def fn_dash(): s = get_stats() top = [[r["content"],r["importance"],r["access_count"]] for r in s["top_memories"]] return s["total_memories"], s["total_knowledge_chunks"], top, s["categories"], s["knowledge_sources"] d_btn.click(fn_dash, outputs=[d_mc,d_kc,d_top,d_cats,d_src]) demo.load(fn_dash, outputs=[d_mc,d_kc,d_top,d_cats,d_src]) # TAB 5 — MCP API with gr.Tab("🔌 MCP API"): gr.Markdown("""### JSON API — Claude Integration Tools: `store_memory` · `search_memories` · `store_knowledge` · `search_knowledge` · `rag_query` · `get_session_context` · `status` ```json {"tool": "store_memory", "params": {"content": "Bytebot uses NestJS + PostgreSQL", "category": "bytebot", "importance": 8}} ```""") api_in = gr.Textbox(label="JSON Request", lines=5, value='{"tool": "status", "params": {}}') api_btn = gr.Button("🚀 Execute", variant="primary") api_out = gr.Textbox(label="Response", lines=14, show_copy_button=True) with gr.Row(): q_st = gr.Button("📊 Status") q_cx = gr.Button("🎯 Session Context") api_btn.click(mcp_api, api_in, api_out) q_st.click(lambda: mcp_api('{"tool":"status","params":{}}'), outputs=api_out) q_cx.click(lambda: mcp_api('{"tool":"get_session_context","params":{}}'), outputs=api_out) demo.launch(server_name="0.0.0.0", server_port=7860, share=False, show_error=True)