app.py

"""
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("""<div class="header">
        <h1 style="color:#60a5fa;font-size:1.9em;margin:0">🔷 VERITAS Memory System</h1>
        <p style="color:#94a3b8;margin:4px 0">CogniVault RAG + Aluna Memory LTM — Unified</p>
        <p style="color:#475569;font-size:0.85em">Rob "The Sounds Guy" Barenbrug &nbsp;|&nbsp; <em>Live in truth, never in comfort</em></p>
    </div>""")

    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)