Update app.py

app.py

@@ -1,4 +1,6 @@
-# JusticeAI Backend — FULL FILE, ALL ENDPOINTS, FIXED /chat, ENSEMBLE LLMS, READY FOR DEPLOYMENT
+# JusticeAI — Full updated app.py
+# Key change: separate knowledge DB (KNOWLEDGEDATABASE_URL) and user DB (DATABASE_URL).
+# /chat now only writes to user_memory (user DB). knowledge DB is only written by /add and background refinement.
 
 import os
 import time
@@ -10,7 +12,7 @@ import asyncio
 import re
 from datetime import datetime, timezone
 from collections import deque
-from typing import Optional, Dict, Any, List
+from typing import Optional, Dict, Any, List, Tuple
 
 import requests
 import psutil
@@ -25,48 +27,62 @@ import logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("justiceai")
 
+# env config
 TRANSLATION_CACHE_DIR = os.environ.get("TRANSLATION_CACHE_DIR", "/tmp/translation_models")
 os.environ["TRANSLATION_CACHE_DIR"] = TRANSLATION_CACHE_DIR
 
 ADMIN_KEY = os.environ.get("ADMIN_KEY")
-DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///justice.db")
+DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///justice.db")  # user DB (user_memory, etc.)
+KNOWLEDGE_DATABASE_URL = os.environ.get("KNOWLEDGEDATABASE_URL", DATABASE_URL)  # knowledge DB (knowledge table)
 EMBED_MODEL_NAME = os.environ.get("EMBED_MODEL_NAME", "paraphrase-multilingual-MiniLM-L12-v2")
 SAVE_MEMORY_CONFIDENCE = float(os.environ.get("SAVE_MEMORY_CONFIDENCE", "0.45"))
 LLM_MODEL_PATHS = [
+    # Examples — replace with local / available checkpoints
     "TinyLlama/TinyLlama-1.1B-Chat-v1.0",
     "Qwen/Qwen1.5-0.5B-Chat",
     "microsoft/phi-2"
 ]
 
+# app + engines
 app = FastAPI(title="JusticeAI — Backend (final)")
-engine = create_engine(
+engine = create_engine(  # user DB (user_memory)
     DATABASE_URL,
     poolclass=NullPool,
     connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
 )
+knowledge_engine = create_engine(  # knowledge DB (knowledge)
+    KNOWLEDGE_DATABASE_URL,
+    poolclass=NullPool,
+    connect_args={"check_same_thread": False} if KNOWLEDGE_DATABASE_URL.startswith("sqlite") else {}
+)
 
-# --- Optional helper imports ---
+# Optional helpers
 try:
     from emojis import get_emoji, get_category_for_mood
 except Exception:
     def get_category_for_mood(mood: str) -> str: return "neutral"
     def get_emoji(cat: str, intensity: float = 0.5) -> str: return "🤖"
+
 try:
     from health import get_health_status
 except Exception:
     def get_health_status(engine_arg) -> Dict[str, Any]: return {"status": "starting", "db_status": "unknown", "stars": 0}
+
 try:
     from langdetect import detect as detect_lang
 except Exception:
     detect_lang = None
+
 try:
     from sentence_transformers import SentenceTransformer
 except Exception:
     SentenceTransformer = None
+
 try:
     from spellchecker import SpellChecker
 except Exception:
     SpellChecker = None
+
 try:
     from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, AutoModelForCausalLM, pipeline as hf_pipeline
 except Exception:
@@ -75,82 +91,91 @@ except Exception:
     AutoModelForCausalLM = None
     hf_pipeline = None
 
-# --- Database creation ---
-def ensure_tables():
-    dialect = engine.dialect.name
-    with engine.begin() as conn:
+# -------------------------
+# Schema setup (both DBs)
+# -------------------------
+def ensure_tables_for_engine(engine_obj, is_knowledge: bool):
+    dialect = engine_obj.dialect.name
+    with engine_obj.begin() as conn:
         if dialect == "sqlite":
-            conn.execute(sql_text("""
-            CREATE TABLE IF NOT EXISTS knowledge (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                text TEXT,
-                reply TEXT,
-                language TEXT DEFAULT 'en',
-                embedding BLOB,
-                category TEXT DEFAULT 'learned',
-                topic TEXT DEFAULT 'general',
-                confidence FLOAT DEFAULT 0,
-                source TEXT,
-                meta TEXT,
-                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            );"""))
-            conn.execute(sql_text("""
-            CREATE TABLE IF NOT EXISTS user_memory (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                user_id TEXT,
-                username TEXT,
-                ip TEXT,
-                text TEXT,
-                reply TEXT,
-                language TEXT DEFAULT 'en',
-                mood TEXT,
-                confidence FLOAT DEFAULT 0,
-                topic TEXT DEFAULT 'general',
-                source TEXT,
-                meta TEXT,
-                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            );"""))
+            if is_knowledge:
+                conn.execute(sql_text("""
+                CREATE TABLE IF NOT EXISTS knowledge (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    text TEXT,
+                    reply TEXT,
+                    language TEXT DEFAULT 'en',
+                    embedding BLOB,
+                    category TEXT DEFAULT 'learned',
+                    topic TEXT DEFAULT 'general',
+                    confidence FLOAT DEFAULT 0,
+                    source TEXT,
+                    meta TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );"""))
+            else:
+                conn.execute(sql_text("""
+                CREATE TABLE IF NOT EXISTS user_memory (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id TEXT,
+                    username TEXT,
+                    ip TEXT,
+                    text TEXT,
+                    reply TEXT,
+                    language TEXT DEFAULT 'en',
+                    mood TEXT,
+                    confidence FLOAT DEFAULT 0,
+                    topic TEXT DEFAULT 'general',
+                    source TEXT,
+                    meta TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );"""))
         else:
-            conn.execute(sql_text("""
-            CREATE TABLE IF NOT EXISTS knowledge (
-                id SERIAL PRIMARY KEY,
-                text TEXT,
-                reply TEXT,
-                language TEXT DEFAULT 'en',
-                embedding BYTEA,
-                category TEXT DEFAULT 'learned',
-                topic TEXT DEFAULT 'general',
-                confidence FLOAT DEFAULT 0,
-                source TEXT,
-                meta JSONB,
-                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            );"""))
-            conn.execute(sql_text("""
-            CREATE TABLE IF NOT EXISTS user_memory (
-                id SERIAL PRIMARY KEY,
-                user_id TEXT,
-                username TEXT,
-                ip TEXT,
-                text TEXT,
-                reply TEXT,
-                language TEXT DEFAULT 'en',
-                mood TEXT,
-                confidence FLOAT DEFAULT 0,
-                topic TEXT DEFAULT 'general',
-                source TEXT,
-                meta JSONB,
-                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            );"""))
-ensure_tables()
-
-def ensure_column_exists(table: str, column: str, col_def_sql: str):
-    dialect = engine.dialect.name
+            if is_knowledge:
+                conn.execute(sql_text("""
+                CREATE TABLE IF NOT EXISTS knowledge (
+                    id SERIAL PRIMARY KEY,
+                    text TEXT,
+                    reply TEXT,
+                    language TEXT DEFAULT 'en',
+                    embedding BYTEA,
+                    category TEXT DEFAULT 'learned',
+                    topic TEXT DEFAULT 'general',
+                    confidence FLOAT DEFAULT 0,
+                    source TEXT,
+                    meta JSONB,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );"""))
+            else:
+                conn.execute(sql_text("""
+                CREATE TABLE IF NOT EXISTS user_memory (
+                    id SERIAL PRIMARY KEY,
+                    user_id TEXT,
+                    username TEXT,
+                    ip TEXT,
+                    text TEXT,
+                    reply TEXT,
+                    language TEXT DEFAULT 'en',
+                    mood TEXT,
+                    confidence FLOAT DEFAULT 0,
+                    topic TEXT DEFAULT 'general',
+                    source TEXT,
+                    meta JSONB,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                );"""))
+
+# Ensure tables exist in both DBs
+ensure_tables_for_engine(knowledge_engine, is_knowledge=True)
+ensure_tables_for_engine(engine, is_knowledge=False)
+
+def ensure_column_exists(engine_obj, table: str, column: str, col_def_sql: str):
+    dialect = engine_obj.dialect.name
     try:
-        with engine.begin() as conn:
+        with engine_obj.begin() as conn:
             if dialect == "sqlite":
                 rows = conn.execute(sql_text(f"PRAGMA table_info({table})")).fetchall()
                 existing_cols = [r[1] for r in rows]
@@ -160,12 +185,17 @@ def ensure_column_exists(table: str, column: str, col_def_sql: str):
                 conn.execute(sql_text(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {col_def_sql}"))
     except Exception:
         pass
-ensure_column_exists("knowledge", "reply", "reply TEXT")
-ensure_column_exists("user_memory", "reply", "reply TEXT")
-ensure_column_exists("knowledge", "language", "language TEXT DEFAULT 'en'")
-ensure_column_exists("knowledge", "embedding", "embedding BYTEA" if engine.dialect.name != "sqlite" else "embedding BLOB")
 
-# --- State ---
+# keep migrations safe
+ensure_column_exists(knowledge_engine, "knowledge", "reply", "reply TEXT")
+ensure_column_exists(knowledge_engine, "knowledge", "language", "language TEXT DEFAULT 'en'")
+ensure_column_exists(knowledge_engine, "knowledge", "embedding", "embedding BYTEA" if knowledge_engine.dialect.name != "sqlite" else "embedding BLOB")
+ensure_column_exists(engine, "user_memory", "reply", "reply TEXT")
+ensure_column_exists(engine, "user_memory", "language", "language TEXT DEFAULT 'en'")
+
+# -------------------------
+# State + telemetry
+# -------------------------
 app_start_time = time.time()
 last_heartbeat = {"time": datetime.utcnow().replace(tzinfo=timezone.utc).isoformat(), "ok": True}
 RECENT_WINDOW_SECONDS = 3600
@@ -175,11 +205,13 @@ recent_requests_timestamps = deque()
 recent_learning_timestamps = deque()
 response_time_ema: Optional[float] = None
 EMA_ALPHA = 0.2
+
 SPARKLINE_LEN = 60
 cpu_history = deque(maxlen=SPARKLINE_LEN)
 mem_history = deque(maxlen=SPARKLINE_LEN)
 latency_history = deque(maxlen=SPARKLINE_LEN)
 recent_metrics = deque(maxlen=600)
+
 model_progress = {
     "embed": {"status": "pending", "progress": 0.0},
     "spell": {"status": "pending", "progress": 0.0},
@@ -190,11 +222,13 @@ model_load_times = {"embed": None, "spell": None, "moderator": None, "llm": None
 embed_model = None
 spell = None
 moderator = None
-ensemble_llms = []
+ensemble_llms: List[Tuple[Any, Any]] = []
 startup_time = 0.0
 _translation_model_cache: Dict[str, Any] = {}
 
-# --- Utility functions ---
+# -------------------------
+# Helpers: text, detection, embedding, LLM ensemble
+# -------------------------
 def record_request(duration_s: float):
     global response_time_ema
     ts = time.time()
@@ -232,7 +266,7 @@ def sanitize_knowledge_text(t: Any) -> str:
         s = s[1:-1].strip()
     return " ".join(s.split())
 
-def dedupe_sentences(text):
+def dedupe_sentences(text: str) -> str:
     parts = re.split(r'([.?!]\s+)', text)
     out = []
     seen = set()
@@ -264,21 +298,16 @@ def detect_language_safe(text: str) -> str:
 def embed_text(text_data: str) -> bytes:
     global embed_model
     if embed_model is None:
-        logger.warning("Embedding model not available; fallback.")
         raise RuntimeError("Embedding model not available.")
-    try:
-        emb = embed_model.encode(text_data, convert_to_tensor=True)
-        return emb.cpu().numpy().tobytes()
-    except Exception as e:
-        logger.warning(f"Embedding fallback: {e}")
-        raise
+    emb = embed_model.encode(text_data, convert_to_tensor=True)
+    return emb.cpu().numpy().tobytes()
 
 def is_boilerplate_candidate(s: str) -> bool:
     s_low = (s or "").strip().lower()
     return "justiceai" in s_low or "dashboard" in s_low or "intelligence" in s_low
 
-def ensemble_llm_suggestions(prompt):
-    replies = []
+def ensemble_llm_suggestions(prompt: str) -> List[str]:
+    replies: List[str] = []
     for tokenizer, model in ensemble_llms:
         try:
             inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512)
@@ -290,6 +319,148 @@ def ensemble_llm_suggestions(prompt):
             logger.debug(f"LLM error ({getattr(tokenizer, 'name_or_path', 'unknown')}): {e}")
     return replies
 
+# -------------------------
+# Synthesis & knowledge utilities (operate on knowledge_engine ONLY)
+# -------------------------
+def create_composite_idea(candidates: List[str], context_prompt: Optional[str] = None) -> str:
+    prompt = "Synthesize a single clear, actionable idea that integrates these proposals:\n\n"
+    for i, c in enumerate(candidates[:8], 1):
+        prompt += f"{i}) {c}\n"
+    if context_prompt:
+        prompt += f"\nContext: {context_prompt}\n"
+    prompt += "\nProduce a concise, integrated plan with benefits and steps."
+    synths = []
+    try:
+        synths = ensemble_llm_suggestions(prompt)
+    except Exception:
+        synths = []
+    if synths:
+        seen = set()
+        merged = []
+        for s in synths:
+            for sent in re.split(r'(?<=[.?!])\s+', s):
+                sent = sent.strip()
+                key = sent.lower()
+                if sent and key not in seen and not is_boilerplate_candidate(sent):
+                    seen.add(key)
+                    merged.append(sent)
+            if len(merged) >= 4:
+                break
+        result = " ".join(merged[:6])
+        if result:
+            return dedupe_sentences(result)
+    reduced = []
+    for c in candidates:
+        c = dedupe_sentences(c)
+        if c and c not in reduced and not is_boilerplate_candidate(c):
+            reduced.append(c)
+    if not reduced:
+        return "I could not synthesize a composite idea; please provide more details."
+    if len(reduced) == 1:
+        return reduced[0]
+    composite = f"Combine: {', '.join(reduced[:3])}."
+    return dedupe_sentences(composite)
+
+def find_similar_knowledge_in_knowledge_db(text: str, topic: str, threshold: float = 0.75) -> Optional[int]:
+    if embed_model is None:
+        return None
+    try:
+        with knowledge_engine.begin() as conn:
+            rows = conn.execute(sql_text("SELECT id, text FROM knowledge WHERE topic = :topic"), {"topic": topic}).fetchall()
+        if not rows:
+            return None
+        ids = [r[0] for r in rows]
+        texts = [r[1] for r in rows]
+        match_embs = embed_model.encode(texts, convert_to_tensor=True)
+        query_emb = embed_model.encode(text, convert_to_tensor=True)
+        sims = torch.nn.functional.cosine_similarity(query_emb.unsqueeze(0), match_embs)
+        best_idx = int(torch.argmax(sims).item())
+        best_score = float(sims[best_idx])
+        if best_score >= threshold:
+            return ids[best_idx]
+    except Exception as e:
+        logger.debug(f"find_similar_knowledge error: {e}")
+        return None
+    return None
+
+def store_or_refine_knowledge_in_knowledge_db(text: str, reply: str, topic: str = "general", confidence: float = 0.5):
+    text = sanitize_knowledge_text(text)
+    reply = sanitize_knowledge_text(reply)
+    try:
+        emb_bytes = None
+        if embed_model is not None:
+            try:
+                emb_bytes = embed_text(text)
+            except Exception:
+                emb_bytes = None
+        existing_id = None
+        try:
+            existing_id = find_similar_knowledge_in_knowledge_db(text, topic, threshold=0.75)
+        except Exception:
+            existing_id = None
+        with knowledge_engine.begin() as conn:
+            if existing_id:
+                # update existing
+                conn.execute(sql_text("""
+                    UPDATE knowledge
+                    SET reply = :reply, text = :text, confidence = GREATEST(coalesce(confidence, 0), :conf), updated_at = CURRENT_TIMESTAMP
+                    WHERE id = :id
+                """), {"reply": reply, "text": text, "conf": float(confidence), "id": existing_id})
+            else:
+                if emb_bytes is not None:
+                    conn.execute(sql_text("""
+                        INSERT INTO knowledge (text, reply, language, embedding, category, topic, confidence)
+                        VALUES (:t, :r, 'en', :e, 'learned', :topic, :conf)
+                    """), {"t": text, "r": reply, "e": emb_bytes, "topic": topic, "conf": float(confidence)})
+                else:
+                    conn.execute(sql_text("""
+                        INSERT INTO knowledge (text, reply, language, category, topic, confidence)
+                        VALUES (:t, :r, 'en', 'learned', :topic, :conf)
+                    """), {"t": text, "r": reply, "topic": topic, "conf": float(confidence)})
+        record_learn_event()
+        return True
+    except Exception as e:
+        logger.warning(f"store_or_refine_knowledge failed: {e}")
+        return False
+
+def deep_refinement_pass():
+    try:
+        with knowledge_engine.begin() as conn:
+            topics_rows = conn.execute(sql_text("SELECT DISTINCT topic FROM knowledge WHERE category='learned'")).fetchall()
+            topics = [r[0] for r in topics_rows if r and r[0]] or ["general"]
+        for t in topics:
+            with knowledge_engine.begin() as conn:
+                rows = conn.execute(sql_text("""
+                    SELECT text, reply, confidence FROM knowledge WHERE topic = :topic AND category='learned'
+                    ORDER BY confidence DESC NULLS LAST, updated_at DESC LIMIT 12
+                """), {"topic": t}).fetchall()
+            candidates = []
+            for r in rows:
+                if r and (r[1] or r[0]):
+                    candidates.append(r[1] or r[0])
+            if not candidates:
+                continue
+            composite = create_composite_idea(candidates, context_prompt=f"topic: {t}")
+            vals = [float(r[2] or 0.0) for r in rows]
+            avg_conf = (sum(vals) / len(vals)) if vals else 0.0
+            composite_conf = min(1.0, avg_conf + 0.15)
+            # store to knowledge DB only
+            store_or_refine_knowledge_in_knowledge_db(composite, composite, topic=t, confidence=composite_conf)
+    except Exception as e:
+        logger.warning(f"deep_refinement_pass error: {e}")
+
+def deep_refinement_loop(interval_minutes: int = 60):
+    while True:
+        try:
+            logger.info("[JusticeAI] Deep refinement tick")
+            deep_refinement_pass()
+        except Exception as e:
+            logger.warning(f"deep_refinement_loop exception: {e}")
+        time.sleep(max(60, interval_minutes * 60))
+
+# -------------------------
+# Startup: load models + background thread
+# -------------------------
 @app.on_event("startup")
 async def startup_event():
     global embed_model, spell, moderator, ensemble_llms, startup_time
@@ -337,6 +508,7 @@ async def startup_event():
         moderator = None
         model_progress["moderator"]["status"] = "error"
         logger.warning(f"[JusticeAI] Moderator load error: {e}")
+
     ensemble_llms.clear()
     if AutoTokenizer is not None and AutoModelForCausalLM is not None:
         for path in LLM_MODEL_PATHS:
@@ -347,22 +519,41 @@ async def startup_event():
                 logger.info(f"[JusticeAI] Loaded ensemble LLM: {path}")
             except Exception as e:
                 logger.warning(f"[JusticeAI] Could not load ensemble LLM {path}: {e}")
+
     startup_time = round(time.time() - t0, 2)
     logger.info(f"[JusticeAI] Startup completed in {startup_time}s")
+
+    # seed some initial knowledge (into knowledge DB only)
     initial_knowledge = [
         {"text": "Justice is fairness in protection of rights and punishment of wrongs.", "reply": "Justice means fairness.", "topic": "general"},
         {"text": "Law is a system of rules created and enforced through social or governmental institutions.", "reply": "Law is a set of rules.", "topic": "general"},
     ]
-    with engine.begin() as conn:
+    with knowledge_engine.begin() as conn:
         for item in initial_knowledge:
             exists = conn.execute(sql_text("SELECT COUNT(*) FROM knowledge WHERE text = :t"), {"t": item["text"]}).scalar()
             if not exists:
-                emb = embed_text(item["text"]) if embed_model else None
-                conn.execute(
-                    sql_text("INSERT INTO knowledge (text, reply, language, embedding, category, topic, confidence) VALUES (:t, :r, 'en', :e, 'learned', :topic, 1.0)"),
-                    {"t": item["text"], "r": item["reply"], "e": emb, "topic": item["topic"]}
-                )
-
+                emb = None
+                if embed_model is not None:
+                    try:
+                        emb = embed_text(item["text"])
+                    except Exception:
+                        emb = None
+                if emb is not None:
+                    conn.execute(sql_text("INSERT INTO knowledge (text, reply, language, embedding, category, topic, confidence) VALUES (:t, :r, 'en', :e, 'learned', :topic, 1.0)"),
+                                 {"t": item["text"], "r": item["reply"], "e": emb, "topic": item["topic"]})
+                else:
+                    conn.execute(sql_text("INSERT INTO knowledge (text, reply, language, category, topic, confidence) VALUES (:t, :r, 'en', 'learned', :topic, 1.0)"),
+                                 {"t": item["text"], "r": item["reply"], "topic": item["topic"]})
+
+    # start deep refinement background thread (runs on knowledge DB)
+    t = threading.Thread(target=deep_refinement_loop, kwargs={"interval_minutes": 60}, daemon=True)
+    t.start()
+
+# -------------------------
+# /chat endpoint — IMPORTANT: only writes to user_memory (engine). DOES NOT write to knowledge.
+# It will use user input to expand internal queries and create composite replies, but will not save
+# those composites directly to knowledge. Knowledge DB is updated only by /add or the background pass.
+# -------------------------
 @app.post("/chat")
 async def chat(request: Request, data: dict = Body(...)):
     t0 = time.time()
@@ -372,36 +563,40 @@ async def chat(request: Request, data: dict = Body(...)):
     username = data.get("username", "anonymous")
     user_ip = request.client.host if request.client else "0.0.0.0"
     user_id = hashlib.sha256(f"{user_ip}-{username}".encode()).hexdigest()
-    topic_hint = str(data.get("topic", "") or "").strip()
+    topic_hint = str(data.get("topic", "") or "").strip() or "general"
     detected_lang = detect_language_safe(raw_msg)
     reply_lang = detected_lang
     user_force_save = bool(data.get("save_memory", False))
 
+    # spell correction
     msg_corrected = raw_msg
     if spell is not None:
         try:
             words = raw_msg.split()
-            corrected = [spell.correction(w) for w in words]
+            corrected = []
+            for w in words:
+                cor = spell.correction(w) if hasattr(spell, "correction") else w
+                corrected.append(cor or w)
             msg_corrected = " ".join(corrected)
         except Exception:
             pass
 
-    topic = topic_hint if topic_hint else "general"
-
+    # retrieve candidates from knowledge DB (read-only)
     try:
-        with engine.begin() as conn:
-            rows = conn.execute(sql_text("SELECT id, text, reply, language, embedding, topic FROM knowledge WHERE category='learned' ORDER BY created_at DESC")).fetchall()
+        with knowledge_engine.begin() as conn:
+            rows = conn.execute(sql_text("SELECT id, text, reply, language, embedding, topic, confidence FROM knowledge WHERE category='learned' ORDER BY confidence DESC, updated_at DESC")).fetchall()
     except Exception as e:
         record_request(time.time() - t0)
         return JSONResponse(status_code=500, content={"error": "failed to read knowledge", "details": str(e)})
 
     knowledge_texts = [r[1] or "" for r in rows]
     knowledge_replies = [r[2] or r[1] or "" for r in rows]
-    knowledge_langs = [r[3] or "en" for r in rows]
     knowledge_topics = [r[5] or "general" for r in rows]
+    knowledge_confidences = [float(r[6] or 0.0) for r in rows]
 
-    matches = []
-    confidence = 0.0
+    # semantic retrieval (local)
+    matches: List[str] = []
+    confidence: float = 0.0
     similarity_threshold = 0.35
     try:
         if embed_model is not None and knowledge_texts:
@@ -409,98 +604,133 @@ async def chat(request: Request, data: dict = Body(...)):
             msg_emb = embed_model.encode(msg_corrected, convert_to_tensor=True)
             if msg_emb.shape[-1] == knowledge_embeddings.shape[-1]:
                 scores = torch.nn.functional.cosine_similarity(msg_emb.unsqueeze(0), knowledge_embeddings)
-                topk = min(10, scores.shape[0])
+                topk = min(12, scores.shape[0])
                 top_indices = torch.topk(scores, k=topk).indices.tolist()
                 seen_text = set()
                 filtered = []
                 for i in top_indices:
                     s = float(scores[i])
                     candidate = knowledge_replies[i]
-                    key = candidate.strip().lower()
+                    key = (candidate or "").strip().lower()
                     if is_boilerplate_candidate(candidate): continue
+                    if not key: continue
                     if key in seen_text: continue
                     seen_text.add(key)
-                    if s > similarity_threshold:
-                        filtered.append((i, s, candidate))
+                    topic_bonus = 0.05 if topic_hint.lower() in (knowledge_topics[i] or "").lower() else 0.0
+                    final_score = s + topic_bonus
+                    if final_score >= similarity_threshold:
+                        filtered.append((i, final_score, candidate))
+                filtered.sort(key=lambda x: x[1], reverse=True)
                 matches = [c for _, _, c in filtered]
                 confidence = filtered[0][1] if filtered else 0.0
             else:
-                logger.warning("Embedding dimension mismatch for confidence scoring.")
+                logger.warning("Embedding dimension mismatch")
                 matches = []
         else:
+            # fallback substring search
             for idx, ktext in enumerate(knowledge_texts):
-                if topic and topic.lower() in (knowledge_topics[idx] or "").lower():
-                    if msg_corrected.lower() in ktext.lower():
-                        matches.append(ktext)
+                if topic_hint and topic_hint.lower() in (knowledge_topics[idx] or "").lower():
+                    if msg_corrected.lower() in ktext.lower() or ktext.lower() in msg_corrected.lower():
+                        matches.append(knowledge_replies[idx])
             confidence = 0.0
     except Exception as e:
-        logger.warning(f"Knowledge retrieval fallback: {e}")
+        logger.warning(f"Retrieval failure: {e}")
         matches = knowledge_replies[:3] if knowledge_replies else []
         confidence = 0.0
 
+    # ask ensemble LLMs for suggestions (non-blocking via executor)
     loop = asyncio.get_running_loop()
-    def run_llm(prompt):
-        return ensemble_llm_suggestions(prompt)
+    def run_llm(prompt_in: str):
+        return ensemble_llm_suggestions(prompt_in)
     try:
-        llm_replies = await loop.run_in_executor(None, run_llm, msg_corrected)
+        prompt_for_llms = f"Respond to: {msg_corrected}\nProvide concise proposals/answers."
+        llm_replies = await loop.run_in_executor(None, run_llm, prompt_for_llms)
     except Exception as e:
         logger.warning(f"LLM ensemble failed: {e}")
         llm_replies = []
 
-    unique_llm_replies = []
+    # dedupe LLMs vs matches (prefer fresh ideas)
+    unique_llm_replies: List[str] = []
     if embed_model is not None and matches and llm_replies:
-        match_embs = embed_model.encode(matches, convert_to_tensor=True)
-        for llm_text in llm_replies:
-            try:
-                llm_emb = embed_model.encode(llm_text, convert_to_tensor=True)
-                sims = torch.nn.functional.cosine_similarity(llm_emb.unsqueeze(0), match_embs)
-                max_sim = float(sims.max().item())
-                if max_sim < 0.60:
-                    unique_llm_replies.append(llm_text)
-            except Exception:
-                if llm_text not in matches:
-                    unique_llm_replies.append(llm_text)
+        try:
+            match_embs = embed_model.encode(matches, convert_to_tensor=True)
+            for llm_text in llm_replies:
+                try:
+                    llm_emb = embed_model.encode(llm_text, convert_to_tensor=True)
+                    sims = torch.nn.functional.cosine_similarity(llm_emb.unsqueeze(0), match_embs)
+                    max_sim = float(sims.max().item())
+                    if max_sim < 0.60:
+                        unique_llm_replies.append(llm_text)
+                except Exception:
+                    if llm_text not in matches:
+                        unique_llm_replies.append(llm_text)
+        except Exception:
+            unique_llm_replies = [r for r in llm_replies if r not in matches]
     else:
-        for llm_text in llm_replies:
-            if llm_text not in matches:
-                unique_llm_replies.append(llm_text)
+        unique_llm_replies = [r for r in llm_replies if r not in matches]
 
+    # combine candidates (knowledge matches + unique LLM replies)
     all_candidates = []
     for m in matches:
         if m and not is_boilerplate_candidate(m):
             all_candidates.append(dedupe_sentences(m))
-    for llm_r in unique_llm_replies:
-        if llm_r and not is_boilerplate_candidate(llm_r):
-            all_candidates.append(dedupe_sentences(llm_r))
+    for l in unique_llm_replies:
+        if l and not is_boilerplate_candidate(l):
+            all_candidates.append(dedupe_sentences(l))
+    # if too few candidates, add user message only as seed (but do not store to knowledge)
+    if not all_candidates:
+        all_candidates.append(msg_corrected)
+
+    # composite idea created (ephemeral). NOTE: do NOT store into knowledge directly here.
+    composite = create_composite_idea(all_candidates, context_prompt=f"topic: {topic_hint}")
+    reply_en = composite if composite else (all_candidates[0] if all_candidates else "I need more details.")
+
+    # ALWAYS: store raw user interaction into user_memory (user DB) — but not to knowledge.
+    try:
+        with engine.begin() as conn:
+            conn.execute(sql_text("""
+                INSERT INTO user_memory (user_id, username, ip, text, reply, language, mood, confidence, topic)
+                VALUES (:uid, :uname, :ip, :text, :reply, :lang, :mood, :conf, :topic)
+            """), {
+                "uid": user_id,
+                "uname": username,
+                "ip": user_ip,
+                "text": raw_msg,
+                "reply": reply_en,
+                "lang": detected_lang,
+                "mood": "neutral",
+                "conf": float(confidence),
+                "topic": topic_hint
+            })
+    except Exception as e:
+        logger.warning(f"/chat user_memory save failed: {e}")
 
-    seen = set()
-    merged = []
-    for s in all_candidates:
-        for sent in re.split(r'(?<=[.?!])\s+', s):
-            sent = sent.strip()
-            if sent and sent not in seen and not is_boilerplate_candidate(sent):
-                seen.add(sent)
-                merged.append(sent)
-    reply_en = " ".join(merged[:3]) if merged else "Can you provide more details so I can help better?"
-
-    reply_final = reply_en
-    mood = "neutral"
-    emoji = ""
-    flags = {}
+    # IMPORTANT: do NOT call store_or_refine_knowledge_in_knowledge_db() here.
+    # The background deep refinement will pick up aggregated data and update knowledge DB.
 
     duration = time.time() - t0
     record_request(duration)
+    emoji = get_emoji(get_category_for_mood("neutral"), intensity=random.random())
+    flags = {}
+
     return {
-        "reply": reply_final,
-        "topic": topic,
+        "reply": reply_en,
+        "topic": topic_hint,
         "language": reply_lang,
         "emoji": emoji,
         "confidence": round(confidence, 2),
         "flags": flags
     }
 
+# -------------------------
+# /add endpoint — explicitly writes to knowledge DB only (admin or trusted)
+# -------------------------
 @app.post("/add")
-async def add_knowledge(data: dict = Body(...)):
+async def add_knowledge(data: dict = Body(...), x_admin_key: str = Header(None, alias="X-Admin-Key")):
+    # optional admin guard; if ADMIN_KEY set and header missing/invalid, deny
+    if ADMIN_KEY:
+        if not x_admin_key or x_admin_key != ADMIN_KEY:
+            return JSONResponse(status_code=403, content={"error": "Invalid or missing admin key."})
     text_data = sanitize_knowledge_text(data.get("text", "") or "")
     reply = sanitize_knowledge_text(data.get("reply", "") or "")
     topic = str(data.get("topic", "") or "").strip()
@@ -508,7 +738,6 @@ async def add_knowledge(data: dict = Body(...)):
         return JSONResponse(status_code=400, content={"error": "Topic is required"})
     if not text_data:
         return JSONResponse(status_code=400, content={"error": "Text is required"})
-    detected = detect_language_safe(text_data)
     try:
         emb = None
         if embed_model is not None:
@@ -517,17 +746,13 @@ async def add_knowledge(data: dict = Body(...)):
             except Exception as e:
                 logger.warning(f"embed_text failed in /add: {e}")
                 emb = None
-        with engine.begin() as conn:
+        with knowledge_engine.begin() as conn:
             if emb is not None:
-                conn.execute(
-                    sql_text("INSERT INTO knowledge (text, reply, language, embedding, category, topic) VALUES (:t, :r, :lang, :e, 'learned', :topic)"),
-                    {"t": text_data, "r": reply, "lang": "en", "e": emb, "topic": topic}
-                )
+                conn.execute(sql_text("INSERT INTO knowledge (text, reply, language, embedding, category, topic) VALUES (:t, :r, :lang, :e, 'learned', :topic)"),
+                             {"t": text_data, "r": reply, "lang": "en", "e": emb, "topic": topic})
             else:
-                conn.execute(
-                    sql_text("INSERT INTO knowledge (text, reply, language, category, topic) VALUES (:t, :r, :lang, 'learned', :topic)"),
-                    {"t": text_data, "r": reply, "lang": "en", "topic": topic}
-                )
+                conn.execute(sql_text("INSERT INTO knowledge (text, reply, language, category, topic) VALUES (:t, :r, :lang, 'learned', :topic)"),
+                             {"t": text_data, "r": reply, "lang": "en", "topic": topic})
         record_learn_event()
         res = {"status": "✅ Knowledge added", "text": text_data, "topic": topic, "language": "en"}
         if embed_model is None or emb is None:
@@ -536,47 +761,14 @@ async def add_knowledge(data: dict = Body(...)):
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": "failed to store knowledge", "details": str(e)})
 
-@app.post("/add-bulk")
-async def add_bulk(data: List[dict] = Body(...)):
-    added = 0
-    errors = []
-    for i, item in enumerate(data):
-        try:
-            text_data = sanitize_knowledge_text(item.get("text", "") or "")
-            reply = sanitize_knowledge_text(item.get("reply", "") or "")
-            topic = str(item.get("topic", "") or "").strip()
-            if not text_data or not topic:
-                errors.append({"index": i, "error": "missing text or topic"})
-                continue
-            detected = detect_language_safe(text_data)
-            try:
-                emb = embed_text(text_data) if embed_model is not None else None
-            except Exception as e:
-                emb = None
-                errors.append({"index": i, "error": f"embed failed: {e}"})
-                continue
-            with engine.begin() as conn:
-                if emb is not None:
-                    conn.execute(
-                        sql_text("INSERT INTO knowledge (text, reply, language, embedding, category, topic) VALUES (:t, :r, :lang, :e, 'learned', :topic)"),
-                        {"t": text_data, "r": reply, "lang": "en", "e": emb, "topic": topic}
-                    )
-                else:
-                    conn.execute(
-                        sql_text("INSERT INTO knowledge (text, reply, language, category, topic) VALUES (:t, :r, :lang, 'learned', :topic)"),
-                        {"t": text_data, "r": reply, "lang": "en", "topic": topic}
-                    )
-            record_learn_event()
-            added += 1
-        except Exception as e:
-            errors.append({"index": i, "error": str(e)})
-    return {"added": added, "errors": errors}
-
+# -------------------------
+# /leaderboard — reads from knowledge DB ONLY
+# -------------------------
 @app.get("/leaderboard")
 async def leaderboard(topic: str = Query("general")):
     topic = str(topic or "general").strip() or "general"
     try:
-        with engine.begin() as conn:
+        with knowledge_engine.begin() as conn:
             rows = conn.execute(sql_text("""
                 SELECT id, text, reply, language, category, confidence, created_at
                 FROM knowledge
@@ -602,6 +794,9 @@ async def leaderboard(topic: str = Query("general")):
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": "failed to fetch leaderboard", "details": str(e)})
 
+# -------------------------
+# model-status, health, metrics (some read both DBs)
+# -------------------------
 @app.get("/model-status")
 async def model_status():
     response_progress = {k: dict(v) for k, v in model_progress.items()}
@@ -617,13 +812,13 @@ async def health_check():
     elapsed = round(time.time() - start, 2)
     health_data["response_time_s"] = elapsed
     try:
-        with engine.connect() as conn:
-            k = conn.execute(sql_text("SELECT COUNT(*) FROM knowledge WHERE category='learned'")).scalar() or 0
-            u = conn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0
+        with knowledge_engine.connect() as kconn, engine.connect() as uconn:
+            k = kconn.execute(sql_text("SELECT COUNT(*) FROM knowledge WHERE category='learned'")).scalar() or 0
+            u = uconn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0
     except Exception:
         k, u = -1, -1
     try:
-        with engine.begin() as conn:
+        with knowledge_engine.begin() as conn:
             rows = conn.execute(sql_text("SELECT DISTINCT topic FROM knowledge WHERE category='learned'")).fetchall()
             topics = [r[0] for r in rows if r and r[0]]
     except Exception:
@@ -638,6 +833,7 @@ async def health_check():
     health_data["learn_rate_per_min"] = sum(1 for t in recent_learning_timestamps if t >= time.time() - 60)
     return health_data
 
+# SSE metrics
 async def metrics_producer():
     while True:
         try:
@@ -654,9 +850,9 @@ async def metrics_producer():
         async def _get_counts():
             def blocking_counts():
                 try:
-                    with engine.connect() as conn:
-                        kcount = conn.execute(sql_text("SELECT COUNT(*) FROM knowledge WHERE category='learned'")).scalar() or 0
-                        ucount = conn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0
+                    with knowledge_engine.connect() as kconn, engine.connect() as uconn:
+                        kcount = kconn.execute(sql_text("SELECT COUNT(*) FROM knowledge WHERE category='learned'")).scalar() or 0
+                        ucount = uconn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0
                     return int(kcount), int(ucount)
                 except Exception:
                     return 0, 0
@@ -701,6 +897,9 @@ async def metrics_recent(limit: int = Query(100, ge=1, le=600)):
     items = list(recent_metrics)[-limit:]
     return {"count": len(items), "metrics": items}
 
+# -------------------------
+# Admin endpoints — operate on knowledge DB for knowledge operations and user DB for user memory operations
+# -------------------------
 @app.post("/verify-admin")
 async def verify_admin(x_admin_key: str = Header(None, alias="X-Admin-Key")):
     if ADMIN_KEY is None:
@@ -719,12 +918,13 @@ async def clear_database(data: dict = Body(...), x_admin_key: str = Header(None,
     if confirm != "CLEAR_DATABASE":
         return JSONResponse(status_code=400, content={"error": "confirm token required."})
     try:
-        with engine.begin() as conn:
-            k_count = conn.execute(sql_text("SELECT COUNT(*) FROM knowledge")).scalar() or 0
-            u_count = conn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0
-            conn.execute(sql_text("DELETE FROM knowledge"))
-            conn.execute(sql_text("DELETE FROM user_memory"))
-        return {"status": "✅ Cleared database", "deleted_knowledge": int(k_count), "deleted_user_memory": int(u_count)}
+        with knowledge_engine.begin() as kconn:
+            k_count = kconn.execute(sql_text("SELECT COUNT(*) FROM knowledge")).scalar() or 0
+            kconn.execute(sql_text("DELETE FROM knowledge"))
+        with engine.begin() as uconn:
+            u_count = uconn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0
+            uconn.execute(sql_text("DELETE FROM user_memory"))
+        return {"status": "✅ Cleared both databases", "deleted_knowledge": int(k_count), "deleted_user_memory": int(u_count)}
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": "failed to clear database", "details": str(e)})
 
@@ -741,7 +941,7 @@ async def reembed_all(data: dict = Body(...), x_admin_key: str = Header(None, al
         return JSONResponse(status_code=400, content={"error": "confirm token required."})
     batch_size = int(data.get("batch_size", 100))
     try:
-        with engine.begin() as conn:
+        with knowledge_engine.begin() as conn:
             rows = conn.execute(sql_text("SELECT id, text FROM knowledge WHERE category='learned' ORDER BY id")).fetchall()
         ids_texts = [(r[0], r[1]) for r in rows]
         total = len(ids_texts)
@@ -752,13 +952,16 @@ async def reembed_all(data: dict = Body(...), x_admin_key: str = Header(None, al
             embs = embed_model.encode(texts, convert_to_tensor=True)
             for j, (kid, _) in enumerate(batch):
                 emb_bytes = embs[j].cpu().numpy().tobytes()
-                with engine.begin() as conn:
+                with knowledge_engine.begin() as conn:
                     conn.execute(sql_text("UPDATE knowledge SET embedding = :e, updated_at = CURRENT_TIMESTAMP WHERE id = :id"), {"e": emb_bytes, "id": kid})
                 updated += 1
         return {"status": "✅ Re-embed complete", "total_rows": total, "updated": updated}
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": "re-embed failed", "details": str(e)})
 
+# -------------------------
+# Frontend dashboard
+# -------------------------
 @app.get("/", response_class=HTMLResponse)
 async def frontend_dashboard():
     try:
@@ -789,6 +992,9 @@ async def frontend_dashboard():
     html = html.replace("%%STARTUP_TIME%%", str(startup_time_local))
     return HTMLResponse(html)
 
+# -------------------------
+# Main
+# -------------------------
 if __name__ == "__main__":
     port = int(os.environ.get("PORT", 7860))
-    uvicorn.run("app:app", host="0.0.0.0", port=port)
+    uvicorn.run("app:app", host="0.0.0.0", port=port)