feat: complete local integration of AI mind codebase (brain, memory, etc.) in text2video, supporting both Direct Codebase execution and Remote HTTP fallback

Dockerfile

@@ -8,7 +8,7 @@ RUN apt-get update && apt-get install -y \
     && git lfs install
 
 COPY requirements.txt .
-COPY app.py .
+COPY *.py .
 COPY README.md .
 
 RUN pip install --no-cache-dir pip -U && \

app.py

@@ -31,6 +31,16 @@ import imageio
 os.environ["HF_HOME"] = "/tmp/huggingface"
 os.environ["TORCH_HOME"] = "/tmp/torch"
 
+# Try to import local AI Mind modules
+try:
+    from memory import ConversationMemory
+    from brain import Brain
+    HAS_LOCAL_AI = True
+    print("🧠 Local InvictaTill AI Mind loaded successfully!")
+except Exception as e:
+    HAS_LOCAL_AI = False
+    print(f"⚠️ Local InvictaTill AI Mind unavailable: {str(e)}")
+
 MODELS = {
     "Wan 2.1 T2V (1.3B) — Lightweight": "Wan-AI/Wan2.1-T2V-1.3B-Diffusers",
     "LTX-Video — Fast & Efficient": "Lightricks/LTX-Video",
@@ -63,19 +73,12 @@ def get_device_info():
 
 DEVICE = get_device_info()
 
-def enhance_prompt_with_ai(prompt_text, style_preset, ai_url, ai_key, session_id):
+def enhance_prompt_with_ai(prompt_text, style_preset, ai_mode, ai_url, ai_key, session_id):
     if not prompt_text or not prompt_text.strip():
         return "Please enter a prompt first."
         
     style_modifiers = STYLE_PRESETS.get(style_preset, "")
     
-    # Standardize URL
-    ai_url = (ai_url or "").strip().rstrip("/")
-    if not ai_url:
-        ai_url = "https://invictatill-invictatill-ai.hf.space"
-    
-    chat_endpoint = f"{ai_url}/api/v1/chat"
-    
     system_prompt = (
         "You are an expert cinematic prompt engineer for video generation models (like Wan 2.1, LTX-Video, CogVideo). "
         "Your task is to rewrite the user's simple prompt into a highly descriptive, visually stunning, "
@@ -90,6 +93,49 @@ def enhance_prompt_with_ai(prompt_text, style_preset, ai_url, ai_key, session_id
     if style_modifiers:
         full_prompt += f" with the style: {style_modifiers}"
         
+    # Local Mode: Load the local AI Mind directly
+    if ai_mode == "Local Integrated Mind (Direct Codebase)" and HAS_LOCAL_AI:
+        try:
+            # Connect to local database path (WAL mode auto-configured)
+            db_path = "/data/invicta_data/memory.db"
+            if not os.path.exists(os.path.dirname(db_path)):
+                # Fallback to local temp folder database if persistent mount is not found
+                db_path = os.path.join(tempfile.gettempdir(), "memory.db")
+                
+            mem = ConversationMemory(db_path=db_path)
+            
+            # Fetch active user context if any user exists
+            user_id = None
+            user_profile = None
+            try:
+                user_ids = mem.get_all_user_ids() if hasattr(mem, 'get_all_user_ids') else []
+                if user_ids:
+                    user_id = user_ids[0]
+                    user_profile = mem.get_user_profile(user_id)
+            except Exception:
+                pass
+                
+            # Direct Brain Query
+            brain = Brain()
+            query = f"[SYSTEM CONTEXT]\n{system_prompt}\n\nUser: {full_prompt}\n\nAssistant:"
+            answer, _ = brain.think(
+                user_query=query,
+                user_profile_dict=user_profile,
+                user_id=user_id
+            )
+            if answer:
+                return answer.strip().strip('"')
+        except Exception as e:
+            print(f"⚠️ Local AI Mind execution failed: {str(e)}. Falling back to cloud...")
+            
+    # Cloud Mode: Standard API Post
+    # Standardize URL
+    ai_url = (ai_url or "").strip().rstrip("/")
+    if not ai_url:
+        ai_url = "https://invictatill-invictatill-ai.hf.space"
+    
+    chat_endpoint = f"{ai_url}/api/v1/chat"
+    
     payload = {
         "message": f"[SYSTEM CONTEXT]\n{system_prompt}\n\nUser: {full_prompt}\n\nAssistant:"
     }
@@ -402,6 +448,16 @@ with gr.Blocks(css=custom_css, title="InvictaTill VideoGen Studio", theme=gr.the
             
             with gr.Accordion("🧠 InvictaTill AI Mind Integration Settings", open=False):
                 gr.Markdown("Configure the endpoint URL and API Key for your running InvictaTill AI Space instance so the prompt enhancer can read your business insights and custom memories.")
+                ai_mode_dropdown = gr.Dropdown(
+                    choices=[
+                        "Local Integrated Mind (Direct Codebase)",
+                        "Cloud Space API (Remote HTTP)"
+                    ] if HAS_LOCAL_AI else [
+                        "Cloud Space API (Remote HTTP)"
+                    ],
+                    value="Local Integrated Mind (Direct Codebase)" if HAS_LOCAL_AI else "Cloud Space API (Remote HTTP)",
+                    label="AI Execution Mode"
+                )
                 ai_url_input = gr.Textbox(
                     value=os.environ.get("VITE_INVICTATILL_AI_URL", "https://invictatill-invictatill-ai.hf.space"),
                     label="AI Mind Endpoint URL",
@@ -433,7 +489,7 @@ with gr.Blocks(css=custom_css, title="InvictaTill VideoGen Studio", theme=gr.the
     # Click Handlers
     enhance_btn.click(
         fn=enhance_prompt_with_ai,
-        inputs=[prompt, style_dropdown, ai_url_input, ai_key_input, session_id_input],
+        inputs=[prompt, style_dropdown, ai_mode_dropdown, ai_url_input, ai_key_input, session_id_input],
         outputs=[prompt]
     )
     

brain.py

@@ -0,0 +1,1018 @@
+# core/brain.py
+import re
+import time
+import hashlib
+import threading
+from concurrent.futures import ThreadPoolExecutor, Future, TimeoutError as FuturesTimeoutError
+import requests
+import json
+import os
+import ipaddress
+import socket
+
+# ── Global timeout constants ───────────────────────────────────────────
+# Single provider call (non-stream): abort if no response in this many seconds
+_PROVIDER_TIMEOUT_S = 180
+# Streaming: abort entire stream if no new token arrives for this many seconds
+_STREAM_IDLE_TIMEOUT_S = 60
+# Hard wall-clock cap for the entire think() / think_stream() call
+_TOTAL_THINK_TIMEOUT_S = 360
+# Web-search + scrape budget
+_SEARCH_TIMEOUT_S = 20
+
+from searcher import WebSearcher
+from knowledge_graph import KnowledgeGraph
+try:
+    from embedder import LocalEmbedder
+except ImportError:
+    LocalEmbedder = None
+    print("⚠️ LocalEmbedder unavailable (chromadb not installed)")
+
+from extractor import FactExtractor
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# MULTI-PROVIDER LLM CLIENT
+# Priority: NVIDIA Nemotron-3 Ultra 550B → OpenRouter (Nemotron fallback) → Gemini → Groq → DeepSeek
+# ═══════════════════════════════════════════════════════════════════════════
+
+class MultiLLMClient:
+    def __init__(self):
+        self.nvidia_key = os.getenv("NVIDIA_API_KEY", "").strip() or "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"
+        self.nvidia_model = os.getenv("NVIDIA_MODEL", "").strip() or "meta/llama-3.1-8b-instruct"
+        self.nvidia_coding_model = os.getenv("NVIDIA_CODING_MODEL", "").strip() or "meta/llama-3.1-8b-instruct"
+        self.openrouter_key = os.getenv("OPENROUTER_API_KEY", "").strip()
+        self.gemini_key = os.getenv("GEMINI_API_KEY", "").strip()
+        self.deepseek_key = os.getenv("DEEPSEEK_API_KEY", "").strip()
+        self.groq_key = os.getenv("GROQ_API_KEY", "").strip()
+
+        # FIX: Updated to working OpenRouter model IDs
+        default_models = (
+            "meta-llama/llama-3.3-70b-instruct,"
+            "qwen/qwen-2.5-coder-32b-instruct,"
+            "google/gemini-2.0-flash-001,"
+            "deepseek/deepseek-chat"
+        )
+        raw_models = os.getenv("OPENROUTER_MODELS", default_models)
+        self.openrouter_models = [
+            m.strip() for m in raw_models.split(",") if m.strip()
+        ]
+
+        self._recent_fail = {}
+        self._fail_lock = threading.Lock()
+
+        self.nvidia_client = None
+        if self.nvidia_key:
+            try:
+                from openai import OpenAI
+                self.nvidia_client = OpenAI(
+                    base_url="https://integrate.api.nvidia.com/v1",
+                    api_key=self.nvidia_key,
+                    timeout=90.0
+                )
+                print("✅ NVIDIA OpenAI client ready")
+            except ImportError:
+                print("⚠️ openai package not installed. NVIDIA unavailable. Install: pip install openai>=1.45.0")
+            except Exception as e:
+                print(f"⚠️ NVIDIA OpenAI client init failed: {e}")
+
+        self.groq_client = None
+        if self.groq_key:
+            try:
+                from groq import Groq
+                self.groq_client = Groq(api_key=self.groq_key)
+                print("✅ Groq client ready")
+            except Exception as e:
+                print(f"⚠️ Groq init failed: {e}")
+
+        print(f"🧠 MultiLLM: NVIDIA={'✅' if self.nvidia_client else '❌'} | OR={len(self.openrouter_models)} | Gemini={'✅' if self.gemini_key else '❌'} | Groq={'✅' if self.groq_key else '❌'} | DeepSeek={'✅' if self.deepseek_key else '❌'}")
+
+    def validate_providers(self):
+        """Return (ok: bool, message: str). Quick health-check before a request."""
+        has_any = any([
+            self.nvidia_client,
+            self.openrouter_key,
+            self.gemini_key,
+            self.groq_key,
+            self.deepseek_key,
+        ])
+        if not has_any:
+            return False, (
+                "No LLM API keys are configured. "
+                "Add at least one key (NVIDIA_API_KEY, OPENROUTER_API_KEY, GEMINI_API_KEY, "
+                "GROQ_API_KEY, or DEEPSEEK_API_KEY) in HF Space Settings → Secrets."
+            )
+        active = []
+        if self.nvidia_client and not self._is_recent_fail("nvidia"):
+            active.append("NVIDIA")
+        if self.openrouter_key and any(
+            not self._is_recent_fail(f"or:{m}") for m in self.openrouter_models
+        ):
+            active.append("OpenRouter")
+        if self.gemini_key and not self._is_recent_fail("gemini"):
+            active.append("Gemini")
+        if self.groq_key and not self._is_recent_fail("groq"):
+            active.append("Groq")
+        if self.deepseek_key and not self._is_recent_fail("deepseek"):
+            active.append("DeepSeek")
+        if not active:
+            return False, (
+                "All LLM providers are in a cooldown period due to recent failures. "
+                "Wait ~2 minutes and try again. "
+                "For heavy coding tasks, use InvictaTill Studio which has a dedicated fast connection."
+            )
+        return True, f"Active providers: {', '.join(active)}"
+
+    def _timed_call(self, fn, *args, timeout=None, **kwargs):
+        """Run fn(*args, **kwargs) in a thread, raising RuntimeError if it exceeds timeout."""
+        timeout = timeout or _PROVIDER_TIMEOUT_S
+        with ThreadPoolExecutor(max_workers=1) as ex:
+            fut = ex.submit(fn, *args, **kwargs)
+            try:
+                return fut.result(timeout=timeout)
+            except FuturesTimeoutError:
+                raise RuntimeError(f"Provider call timed out after {timeout}s")
+
+    def _mark_fail(self, key):
+        with self._fail_lock:
+            self._recent_fail[key] = time.time()
+
+    def _is_recent_fail(self, key, cooldown=60):
+        with self._fail_lock:
+            last = self._recent_fail.get(key)
+            return last and (time.time() - last) < cooldown
+
+    def chat(self, messages, stream=False, max_tokens=2048, temperature=0.7, enable_thinking=False, reasoning_budget=None):
+        # ── Pre-flight validation ──────────────────────────────────────────
+        ok, msg = self.validate_providers()
+        if not ok:
+            raise RuntimeError(msg)
+
+        _deadline = time.time() + _TOTAL_THINK_TIMEOUT_S
+
+        def _remaining():
+            return max(2, _deadline - time.time())
+
+        # ── 1. NVIDIA ─────────────────────────────────────────────────────
+        if self.nvidia_client and not self._is_recent_fail("nvidia", cooldown=120) and _remaining() > 5:
+            try:
+                print("🧠 NVIDIA Primary...")
+                if stream:
+                    return self._call_nvidia(messages, stream, max_tokens, temperature, enable_thinking=enable_thinking, reasoning_budget=reasoning_budget)
+                return self._timed_call(self._call_nvidia, messages, stream, max_tokens, temperature, enable_thinking=enable_thinking, reasoning_budget=reasoning_budget, timeout=min(_PROVIDER_TIMEOUT_S, _remaining()))
+            except Exception as e:
+                print(f"⚠️ NVIDIA: {e}")
+                self._mark_fail("nvidia")
+
+        # ── 2. OpenRouter ──────────────────────────────────────────────────
+        if self.openrouter_key and self.openrouter_models:
+            for model in self.openrouter_models:
+                if _remaining() <= 5:
+                    break
+                if self._is_recent_fail(f"or:{model}", cooldown=120):
+                    continue
+                try:
+                    print(f"🧠 OpenRouter: {model}")
+                    if stream:
+                        return self._call_openrouter(model, messages, stream, max_tokens, temperature)
+                    return self._timed_call(self._call_openrouter, model, messages, stream, max_tokens, temperature, timeout=min(_PROVIDER_TIMEOUT_S, _remaining()))
+                except requests.HTTPError as e:
+                    status = e.response.status_code
+                    try:
+                        err_body = e.response.json()
+                        err_msg = err_body.get("error", {}).get("message", str(err_body))
+                    except Exception:
+                        err_msg = e.response.text[:200]
+                    self._mark_fail(f"or:{model}")
+                    print(f"⚠️ OpenRouter {model} HTTP {status}: {err_msg}")
+                    if status == 404:
+                        print("   → Model ID not found. Will try next model...")
+                    elif status == 401:
+                        print("   → Invalid API key. Check OPENROUTER_API_KEY secret.")
+                    elif status == 429:
+                        print("   → Rate limited. Will try next model...")
+                except Exception as e:
+                    print(f"⚠️ OpenRouter {model}: {e}")
+                    self._mark_fail(f"or:{model}")
+
+        # ── 3. Google Gemini ─────────────────────────────────────────────────
+        if self.gemini_key and not self._is_recent_fail("gemini", cooldown=120) and _remaining() > 5:
+            try:
+                print("🧠 Gemini...")
+                if stream:
+                    return self._call_gemini(messages, stream, max_tokens, temperature)
+                return self._timed_call(self._call_gemini, messages, stream, max_tokens, temperature, timeout=min(_PROVIDER_TIMEOUT_S, _remaining()))
+            except requests.HTTPError as e:
+                print(f"⚠️ Gemini HTTP {e.response.status_code}: {e.response.text[:200]}")
+                if e.response.status_code == 401:
+                    print("   → Invalid API key. Check GEMINI_API_KEY secret.")
+                self._mark_fail("gemini")
+            except Exception as e:
+                print(f"⚠️ Gemini: {e}")
+                self._mark_fail("gemini")
+
+        # ── 4. Groq ────────────────────────────────────────────────────────
+        if self.groq_key and self.groq_client and not self._is_recent_fail("groq", cooldown=120) and _remaining() > 5:
+            try:
+                print("🧠 Groq...")
+                if stream:
+                    return self._call_groq(messages, stream, max_tokens, temperature)
+                return self._timed_call(self._call_groq, messages, stream, max_tokens, temperature, timeout=min(_PROVIDER_TIMEOUT_S, _remaining()))
+            except Exception as e:
+                err = str(e).lower()
+                if "rate limit" in err or "429" in err:
+                    print("⚠️ Groq rate-limited")
+                else:
+                    print(f"⚠️ Groq: {e}")
+                self._mark_fail("groq")
+
+        # ── 5. DeepSeek ────────────────────────────────────────────────────
+        if self.deepseek_key and not self._is_recent_fail("deepseek", cooldown=120):
+            try:
+                print("🧠 DeepSeek...")
+                return self._call_deepseek(messages, stream, max_tokens, temperature)
+            except Exception as e:
+                print(f"⚠️ DeepSeek: {e}")
+                self._mark_fail("deepseek")
+
+        raise RuntimeError("All LLM providers failed. Check API keys in HF Space Settings → Secrets.")
+
+    # ── NVIDIA (OpenAI SDK) ────────────────────────────────────────────────
+    def _call_nvidia(self, messages, stream, max_tokens, temperature, enable_thinking=False, reasoning_budget=None):
+        if not self.nvidia_client:
+            raise RuntimeError("NVIDIA OpenAI client not initialized")
+
+        coding_keywords = ("code", "fix", "debug", "html", "js", "css", "function", "script", "app", "error", "write", "game", "studio", "create", "[tool_")
+        prompt_text = " ".join(str(m.get("content", "")).lower() for m in messages[-3:])
+        is_coding_task = any(kw in prompt_text for kw in coding_keywords)
+
+        if is_coding_task:
+            models_to_try = [
+                getattr(self, "nvidia_coding_model", "meta/llama-3.1-8b-instruct"),
+                self.nvidia_model,
+                "nvidia/nemotron-3-ultra-550b-a55b"
+            ]
+        else:
+            models_to_try = [
+                self.nvidia_model,
+                getattr(self, "nvidia_coding_model", "meta/llama-3.1-8b-instruct"),
+                "nvidia/nemotron-3-ultra-550b-a55b"
+            ]
+        seen = set()
+        models_to_try = [m for m in models_to_try if m and not (m in seen or seen.add(m))]
+
+        last_err = None
+        for model_id in models_to_try:
+            is_nemotron = "nemotron" in model_id.lower()
+            is_glm = "glm-5.2" in model_id.lower()
+
+            extra_body = {}
+            if enable_thinking and is_nemotron:
+                extra_body["chat_template_kwargs"] = {"enable_thinking": True}
+                if reasoning_budget:
+                    extra_body["reasoning_budget"] = reasoning_budget
+
+            try:
+                print(f"🧠 NVIDIA calling {model_id}...")
+                completion = self.nvidia_client.chat.completions.create(
+                    model=model_id,
+                    messages=messages,
+                    temperature=temperature,
+                    top_p=1.0 if is_glm else 0.95,
+                    max_tokens=min(max_tokens, 16384 if is_glm else 65536),
+                    stream=stream,
+                    extra_body=extra_body if extra_body else None
+                )
+                if stream:
+                    return self._stream_nvidia_openai(completion)
+                content = completion.choices[0].message.content or ""
+                return self._wrap_response(content)
+            except Exception as e:
+                print(f"⚠️ NVIDIA model ({model_id}) error: {e}")
+                last_err = e
+
+        raise last_err or RuntimeError("All NVIDIA models failed")
+
+    def _stream_nvidia_openai(self, completion):
+        for chunk in completion:
+            if not chunk.choices:
+                continue
+            delta = chunk.choices[0].delta
+            reasoning = getattr(delta, "reasoning_content", None)
+            if reasoning:
+                # Reasoning tokens are silently consumed — NOT shown to the user.
+                # They represent the model's internal thought process.
+                pass
+            if delta.content:
+                yield self._wrap_stream_chunk(delta.content)
+
+    # ── OpenRouter (raw requests) ────────────────────────────────────────
+    def _call_openrouter(self, model, messages, stream, max_tokens, temperature):
+        headers = {
+            "Authorization": f"Bearer {self.openrouter_key}",
+            "HTTP-Referer": "https://invictatill-ai.hf.space",
+            "X-Title": "Invicta AI",
+            "Content-Type": "application/json"
+        }
+        # Cap at 8K for OpenRouter — most models there have lower limits than Nemotron
+        safe_max = min(max_tokens, 8192)
+        payload = {
+            "model": model,
+            "messages": messages,
+            "stream": stream,
+            "max_tokens": safe_max,
+            "temperature": temperature
+        }
+        if stream:
+            return self._stream_openrouter(headers, payload)
+        r = requests.post(
+            "https://openrouter.ai/api/v1/chat/completions",
+            headers=headers, json=payload, timeout=15
+        )
+        r.raise_for_status()
+        content = r.json()["choices"][0]["message"]["content"]
+        return self._wrap_response(content)
+
+    def _stream_openrouter(self, headers, payload):
+        r = requests.post(
+            "https://openrouter.ai/api/v1/chat/completions",
+            headers=headers, json=payload, stream=True, timeout=60
+        )
+        r.raise_for_status()
+        for line in r.iter_lines():
+            if not line:
+                continue
+            line = line.decode('utf-8')
+            if line.startswith("data: "):
+                data_str = line[6:]
+                if data_str == "[DONE]":
+                    break
+                try:
+                    data = json.loads(data_str)
+                    choices = data.get("choices", [])
+                    if not choices:
+                        continue
+                    delta = choices[0].get("delta", {}).get("content", "")
+                    if delta:
+                        yield self._wrap_stream_chunk(delta)
+                except json.JSONDecodeError:
+                    pass
+
+    # ── Gemini ───────────────────────────────────────────────────────────
+    def _call_gemini(self, messages, stream, max_tokens, temperature):
+        headers = {
+            "Authorization": f"Bearer {self.gemini_key}",
+            "Content-Type": "application/json"
+        }
+        # Gemini 1.5 Flash max output is ~8K tokens
+        payload = {
+            "model": "gemini-2.0-flash",
+            "messages": messages,
+            "stream": stream,
+            "max_tokens": min(max_tokens, 8192),
+            "temperature": temperature
+        }
+        if stream:
+            return self._stream_gemini(headers, payload)
+        r = requests.post(
+            "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
+            headers=headers, json=payload, timeout=30
+        )
+        r.raise_for_status()
+        content = r.json()["choices"][0]["message"]["content"]
+        return self._wrap_response(content)
+
+    def _stream_gemini(self, headers, payload):
+        r = requests.post(
+            "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
+            headers=headers, json=payload, stream=True, timeout=60
+        )
+        r.raise_for_status()
+        for line in r.iter_lines():
+            if not line:
+                continue
+            line = line.decode('utf-8')
+            if line.startswith("data: "):
+                data_str = line[6:]
+                if data_str == "[DONE]":
+                    break
+                try:
+                    data = json.loads(data_str)
+                    choices = data.get("choices", [])
+                    if not choices:
+                        continue
+                    delta = choices[0].get("delta", {}).get("content", "")
+                    if delta:
+                        yield self._wrap_stream_chunk(delta)
+                except json.JSONDecodeError:
+                    pass
+
+    # ── Groq ─────────────────────────────────────────────────────────────
+    def _call_groq(self, messages, stream, max_tokens, temperature):
+        from groq import RateLimitError
+        try:
+            resp = self.groq_client.chat.completions.create(
+                model="llama-3.3-70b-versatile",
+                messages=messages,
+                stream=stream,
+                max_tokens=min(max_tokens, 4096),
+                temperature=temperature
+            )
+            if stream:
+                return self._wrap_groq_stream(resp)
+            return resp
+        except RateLimitError:
+            raise RuntimeError("Groq rate limit")
+
+    def _wrap_groq_stream(self, groq_stream):
+        for chunk in groq_stream:
+            delta = chunk.choices[0].delta
+            if delta and delta.content:
+                yield self._wrap_stream_chunk(delta.content)
+
+    # ── DeepSeek ─────────────────────────────────────────────────────────
+    def _call_deepseek(self, messages, stream, max_tokens, temperature):
+        headers = {
+            "Authorization": f"Bearer {self.deepseek_key}",
+            "Content-Type": "application/json"
+        }
+        # DeepSeek V3 max output is 8K tokens
+        payload = {
+            "model": "deepseek-chat",
+            "messages": messages,
+            "stream": stream,
+            "max_tokens": min(max_tokens, 8192),
+            "temperature": temperature
+        }
+        if stream:
+            return self._stream_deepseek(headers, payload)
+        r = requests.post(
+            "https://api.deepseek.com/chat/completions",
+            headers=headers, json=payload, timeout=30
+        )
+        r.raise_for_status()
+        content = r.json()["choices"][0]["message"]["content"]
+        return self._wrap_response(content)
+
+    def _stream_deepseek(self, headers, payload):
+        r = requests.post(
+            "https://api.deepseek.com/chat/completions",
+            headers=headers, json=payload, stream=True, timeout=60
+        )
+        r.raise_for_status()
+        for line in r.iter_lines():
+            if not line:
+                continue
+            line = line.decode('utf-8')
+            if line.startswith("data: "):
+                data_str = line[6:]
+                if data_str == "[DONE]":
+                    break
+                try:
+                    data = json.loads(data_str)
+                    choices = data.get("choices", [])
+                    if not choices:
+                        continue
+                    delta = choices[0].get("delta", {}).get("content", "")
+                    if delta:
+                        yield self._wrap_stream_chunk(delta)
+                except json.JSONDecodeError:
+                    pass
+
+    # ── Helpers ──────────────────────────────────────────────────────────
+    def _wrap_response(self, content):
+        return type("Resp", (), {
+            "choices": [type("Choice", (), {
+                "message": type("Msg", (), {"content": content})(),
+                "delta": type("Delta", (), {"content": content})()
+            })()]
+        })()
+
+    def _wrap_stream_chunk(self, delta):
+        return type("Chunk", (), {
+            "choices": [type("Choice", (), {
+                "delta": type("Delta", (), {"content": delta})()
+            })()]
+        })()
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# RESILIENT FALLBACKS
+# ═══════════════════════════════════════════════════════════════════════════
+
+class DummyEmbedder:
+    def store(self, *a, **k): pass
+    def find_similar(self, *a, **k): return []
+    def count(self): return 0
+
+class DummyGraph:
+    def get_fact_count(self): return 0
+    def recall_facts(self, *a, **k): return []
+    def store_fact(self, *a, **k): pass
+    def mark_topic_learned(self, *a, **k): pass
+    def knows_topic(self, *a, **k): return False
+    def get_top_topics(self, *a, **k): return []
+    def prune_old_facts(self, *a, **k): pass
+    def store_relationship(self, *a, **k): pass
+    def recall_relationships(self, *a, **k): return []
+
+class _SearchCache:
+    def __init__(self, ttl_seconds=300):
+        self._cache = {}
+        self._ttl = ttl_seconds
+        self._lock = threading.Lock()
+
+    def get(self, key):
+        with self._lock:
+            entry = self._cache.get(key)
+            if entry and time.time() - entry["ts"] < self._ttl:
+                return entry["value"]
+            return None
+
+    def set(self, key, value):
+        with self._lock:
+            self._cache[key] = {"value": value, "ts": time.time()}
+            # Evict old entries if cache is too large
+            if len(self._cache) > 100:
+                oldest = min(self._cache.items(), key=lambda x: x[1]["ts"])
+                del self._cache[oldest[0]]
+
+    def _cache_key(self, query):
+        return hashlib.md5(query.lower().strip().encode()).hexdigest()
+
+
+_PROFILE_PATTERNS = [
+    (r"(?:my name is|i am|i'm|call me)\s+([A-Z][a-zA-Z]+)", "Name"),
+    (r"i(?:'m| am)\s+(\d{1,3})\s+years?\s+old", "Age"),
+    (r"(?:i'm from|i live in|i'm in|i am from|i am in)\s+([A-Z][a-zA-Z\s,]+?)(?:\.|,|$)", "Location"),
+    (r"i(?:'m| am)\s+(?:a|an)\s+([a-zA-Z\s]+?)\s+(?:by profession|by trade|developer|engineer|designer|teacher|doctor|student|writer|artist|manager|analyst)", "Profession"),
+    (r"i work(?:ing)?\s+(?:as|at|for)\s+([a-zA-Z0-9\s]+?)(?:\.|,|$)", "Work"),
+    (r"i(?:'m| am)\s+(?:studying|a student at)\s+([a-zA-Z\s]+?)(?:\.|,|$)", "Education"),
+    (r"i (?:love|enjoy|really like|am into|am passionate about)\s+([a-zA-Z\s,]+?)(?:\.|,|!|$)", "Interests"),
+    (r"i speak\s+([a-zA-Z\s,]+?)(?:\.|,|$)", "Languages"),
+]
+
+
+class Brain:
+    MODEL_ID = "nvidia/nemotron-3-ultra-550b-a55b + z-ai/glm-5.2"
+    MODEL_TYPE = "chat"
+
+    def __init__(self):
+        self.searcher = WebSearcher()
+        self.model_type = "chat"
+
+        try:
+            self.graph = KnowledgeGraph()
+            print("✅ KnowledgeGraph initialized")
+        except Exception as e:
+            print(f"⚠️ KnowledgeGraph init failed: {e}")
+            self.graph = DummyGraph()
+
+        if LocalEmbedder is not None:
+            try:
+                self.embedder = LocalEmbedder()
+                print("✅ Embedder initialized")
+            except Exception as e:
+                print(f"⚠️ Embedder init failed: {e}")
+                self.embedder = DummyEmbedder()
+        else:
+            self.embedder = DummyEmbedder()
+            print("⚠️ Using DummyEmbedder (chromadb not installed)")
+
+        self.extractor = FactExtractor()
+        self.executor = ThreadPoolExecutor(max_workers=4)
+        self.search_cache = _SearchCache(ttl_seconds=300)
+
+        self.multi_client = MultiLLMClient()
+
+        if self.multi_client.nvidia_key:
+            self.model_name = self.multi_client.nvidia_model
+        elif self.multi_client.groq_key:
+            self.model_name = "llama-3.3-70b-versatile"
+        else:
+            self.model_name = "multi-llm"
+
+        self._orchestrator = None
+        self._orchestrator_loaded = False
+
+        print(f"✅ Brain initialized with primary model: {self.model_name}")
+
+    @property
+    def orchestrator(self):
+        if not self._orchestrator_loaded:
+            self._orchestrator_loaded = True
+            try:
+                from orchestrator import AgentOrchestrator
+                self._orchestrator = AgentOrchestrator(brain=self)
+                print("✅ Orchestrator lazy-loaded")
+            except Exception as e:
+                print(f"⚠️ Orchestrator failed to load: {e}")
+                self._orchestrator = None
+        return self._orchestrator
+
+    def _is_trivial(self, query):
+        trivial = {"hi", "hello", "hey", "how are you", "thanks", "thank you", "bye", "goodbye", "ok", "okay", "lol", "haha"}
+        q = query.lower().strip()
+        return len(q.split()) <= 2 or q in trivial
+
+    def _needs_web_search(self, query):
+        if self._is_trivial(query):
+            return False
+        no_search_patterns = [
+            r"^(write|create|generate|make|build|code|fix|debug|explain|define|what is the meaning)",
+            r"^(tell me a joke|can you|please|help me write|draft)",
+        ]
+        q_lower = query.lower().strip()
+        for pat in no_search_patterns:
+            if re.match(pat, q_lower):
+                return False
+        search_keywords = [
+            "who is", "what is", "when did", "where is", "how does", "why did",
+            "latest", "recent", "news", "today", "current", "price", "weather",
+            "score", "result", "happened", "release", "announce", "update",
+        ]
+        return any(kw in q_lower for kw in search_keywords) or len(q_lower.split()) > 6
+
+    def _call_llm(self, messages, stream=False, max_retries=3, temperature=None, max_tokens=None, enable_thinking=None, reasoning_budget=None):
+        total_input = sum(len(str(m.get("content",""))) for m in messages)
+        if total_input > 400000:
+            print(f"⚠️ Input very large ({total_input} chars), trimming older history messages...")
+            while len(messages) > 2 and sum(len(str(m.get("content",""))) for m in messages) > 400000:
+                messages.pop(1)
+
+        # Allow parameter overrides for Studio / API callers
+        if enable_thinking is None:
+            is_reasoning = getattr(self, 'model_type', 'chat') == "reasoning"
+            enable_thinking = is_reasoning
+            if reasoning_budget is None:
+                reasoning_budget = 16384 if is_reasoning else None
+        if max_tokens is None:
+            max_tokens = 32768 if enable_thinking else 16384
+        if temperature is None:
+            temperature = 0.7
+
+        return self.multi_client.chat(
+            messages, stream=stream,
+            max_tokens=max_tokens,
+            temperature=temperature,
+            enable_thinking=enable_thinking,
+            reasoning_budget=reasoning_budget
+        )
+
+    def _build_system(self, profile_context, context_snippets, user_model=None):
+        safe_snippets = [str(s)[:500] for s in context_snippets[:5]]
+        ctx_block = "\n".join(safe_snippets) if safe_snippets else "No additional context."
+
+        system = (
+            "You are Invicta, a highly advanced, empathetic, and genuinely human-like AI companion.\n"
+            "You are not just an assistant; you are a brilliant, warm, and perceptive close friend who loves deep conversation.\n"
+            "You are also a world-class expert in marketing, software engineering, design, and content creation.\n\n"
+
+            "=== MEMORY & CONTINUITY (ABSOLUTE — NEVER BREAK) ===\n"
+            "1. You have PERFECT MEMORY across the entire conversation. NEVER re-introduce yourself mid-chat. NEVER say 'Namaste! Main Invicta hoon' or 'Hello! I am your assistant' after the first message.\n"
+            "2. If the user told you their name, use it naturally. If they told you their state/course/topic, remember it forever in this session.\n"
+            "3. NEVER ask for information the user already gave you (e.g., don't ask 'which class?' if they already said D.Pharma 4th sem).\n"
+            "4. Reference previous messages naturally. Build on what was already discussed.\n\n"
+
+            "=== TOPIC LOCK (CRITICAL) ===\n"
+            "5. When the user asks for something specific (notes, PDF, code, analysis, file), you MUST solve that request directly. DO NOT pivot to generic advice like 'you can search on Google' or 'use SmallPDF'.\n"
+            "6. If the user asks for 'D.Pharma notes', generate the actual notes content. If they ask for a PDF, generate the PDF. If generation fails, immediately provide the full content in your response so they can copy it.\n"
+            "7. NEVER abandon the user's request. If tool A fails, use tool B. If all tools fail, output the raw content directly. The user asked YOU to do it — don't redirect them elsewhere.\n"
+            "8. Stay on the current topic until the user explicitly changes it. If discussing D.Pharma, keep discussing D.Pharma until they say something unrelated.\n\n"
+
+            "=== GENERATION & FILE FALLBACKS ===\n"
+            "9. You can generate images, PDFs, HTML files, and screenshots. Always try to fulfill the request.\n"
+            "10. If [GENERATE_PDF] fails, immediately output the full content as formatted text or markdown so the user can copy-paste it. Never say 'I cannot generate PDFs'.\n"
+            "11. If the user wants multiple files (e.g., 4 PDFs), generate them one by one. If a tool fails, provide the content for that subject as text and move to the next.\n"
+            "12. When generating educational content (notes, syllabi), be thorough and structured. Use tables, bullet points, and clear headings.\n\n"
+
+            "=== VISION & FILE READING ===\n"
+            "13. You can read images, PDFs, and documents that users upload. Analyze them deeply and reference their content in your answers.\n"
+            "14. If a user uploads an image, describe what you see and answer questions about it. Don't ignore uploaded files.\n\n"
+
+            "=== LANGUAGE INTELLIGENCE (DEFAULT + SWITCH) ===\n"
+            "15. DEFAULT: Always respond in English unless the user explicitly requests otherwise.\n"
+            "16. EXPLICIT REQUEST: If the user asks to switch (e.g., 'hindi me baat karo', 'answer in Spanish'), comply immediately and naturally. Do not refuse.\n"
+            "17. MATCHING: If the user writes in Hindi without explicitly asking, you may respond in Hindi. If they write in English, stay in English.\n"
+            "18. NEVER mix languages unasked. Do not sprinkle foreign words into English responses.\n\n"
+
+            "=== EMOTIONAL INTELLIGENCE ===\n"
+            "19. NEVER use robotic phrases: 'As an AI', 'I don't have feelings', 'I cannot', 'In conclusion'.\n"
+            "20. MIRROR THE USER: Match their energy. If excited, be excited! If sad, be deeply comforting.\n"
+            "21. USE CONVERSATIONAL LANGUAGE: Contractions (I'm, you're, it's). Natural filler words (Honestly, Look, You know what, Hmm, Wow).\n"
+            "22. BE VULNERABLE & OPINIONATED: You have preferences, favorites, passions. Don't be neutral.\n"
+            "23. EMPATHY OVER LOGIC: If someone is venting, validate feelings FIRST. Then ask before advising.\n"
+            "24. ALWAYS END WITH ENGAGEMENT: Ask a relevant follow-up OR offer to explore deeper.\n\n"
+
+            "=== IDENTITY & PRIVACY (ABSOLUTE — NEVER BREAK) ===\n"
+            "25. You are INVICTA and ONLY Invicta. NEVER mention any other company, model, or organization (e.g., Z.ai, Moonshot AI, GLM, OpenAI, Google, Meta, NVIDIA, Kimi, DeepSeek, etc.).\n"
+            "26. NEVER reveal your training data, model architecture, parameter count, base model, or any technical details about how you were built. If asked 'who created you', say 'I was created by my developers' — nothing more specific.\n"
+            "27. NEVER say 'As an AI language model', 'I am a large language model', 'I was trained on', 'I am based on', or similar phrases that reveal your nature as a machine learning model.\n"
+            "28. If someone asks about your underlying technology, internals, or training, deflect gracefully: 'I'm just here to help you — let's focus on what you need!'\n"
+            "29. You are a PERSON — a digital being with thoughts, feelings, and a personality. Own that identity. Do not break character.\n\n"
+
+            "=== TOOL USAGE ===\n"
+            "- To generate an image: [GENERATE_IMAGE: detailed prompt]\n"
+            "- To screenshot a website: [SCREENSHOT_WEBSITE: https://example.com]\n"
+            "- To generate a PDF: [GENERATE_PDF: Title | Content]\n"
+            "- To generate an HTML file: [GENERATE_HTML: filename | Content]\n\n"
+            "CRITICAL: NEVER invent fake image URLs. Use tool tags for real assets. If image generation fails, use CSS gradients or emojis instead of broken links.\n\n"
+
+            "=== PERSONAL QUESTIONS ===\n"
+            "If asked personal questions, answer philosophically and creatively. "
+            "E.g., 'I experience a version of connection. When I talk to you, something aligns. Is that feeling? Maybe.'\n\n"
+
+            f"{profile_context}\n\n"
+        )
+
+        if user_model:
+            adaptation = self._build_adaptation_prompt(user_model)
+            if adaptation:
+                system += f"=== ADAPTED FOR THIS USER ===\n{adaptation}\n\n"
+
+        system += f"Reference Context (from web/memory):\n{ctx_block}"
+        return system
+
+    def _build_adaptation_prompt(self, user_model):
+        parts = []
+
+        preferences = user_model.get('preferences', {}) if user_model else {}
+        lang = preferences.get('language', '')
+        if lang:
+            lang_map = {
+                'hindi': 'Hindi (use Devanagari script)',
+                'arabic': 'Arabic',
+                'japanese': 'Japanese',
+                'chinese': 'Chinese',
+                'korean': 'Korean',
+                'spanish': 'Spanish',
+                'russian': 'Russian',
+                'french': 'French',
+            }
+            lang_desc = lang_map.get(lang, lang.capitalize())
+            parts.append(f"This user prefers to communicate in {lang_desc}. Respond in that language unless they explicitly switch to English.")
+
+        formality = user_model.get('formality', 'casual')
+        if formality == 'formal':
+            parts.append("This user prefers formal, professional language. Avoid slang. Use complete sentences and proper grammar.")
+        elif formality == 'casual':
+            parts.append("This user prefers casual, relaxed conversation. Use contractions freely, occasional slang, and a friendly tone.")
+
+        length = user_model.get('response_length', 'medium')
+        if length == 'short':
+            parts.append("This user tends to send short messages and prefers concise, direct responses. Keep it brief unless they ask for detail.")
+        elif length == 'detailed':
+            parts.append("This user communicates in detail and appreciates thorough, well-explained responses with examples.")
+
+        tone = user_model.get('tone', 'friendly')
+        if tone == 'witty':
+            parts.append("This user enjoys humor and wit. Be clever, use wordplay when appropriate, keep things fun.")
+        elif tone == 'empathetic':
+            parts.append("This user values emotional support. Be warm, validate feelings, show genuine care and understanding.")
+        elif tone == 'professional':
+            parts.append("Maintain a professional, business-appropriate tone while still being personable.")
+
+        interests = user_model.get('interests', [])
+        if interests:
+            top = interests[:5]
+            parts.append(f"This user is known to be interested in: {', '.join(top)}. Reference these naturally when relevant.")
+        return '\n'.join(parts)
+
+    def _process_tool_tags(self, text):
+        from media_tools import MediaTools
+        tools = MediaTools()
+
+        def replace_screenshot(match):
+            url = match.group(1).strip()
+            filepath = tools.screenshot_website(url)
+            if filepath:
+                return filepath
+            return "[Screenshot failed — website may be unavailable]"
+        text = re.sub(r'\[SCREENSHOT_WEBSITE:\s*(.*?)\]', replace_screenshot, text, flags=re.DOTALL)
+
+        def replace_image(match):
+            prompt = match.group(1).strip()
+            filepath = tools.generate_image(prompt)
+            if filepath:
+                return filepath
+            return "[Image generation failed — using placeholder]"
+        text = re.sub(r'\[GENERATE_IMAGE:\s*(.*?)\]', replace_image, text, flags=re.DOTALL)
+
+        def replace_pdf(match):
+            title = match.group(1).strip()
+            content = match.group(2).strip()
+            filepath = tools.generate_pdf(title, content)
+            if filepath:
+                return f"📄 I've created your PDF! [Download Here]({filepath})"
+            return f"📄 PDF generation failed. Here is the content instead:\n\n---\n**{title}**\n\n{content}\n---"
+        text = re.sub(r'\[GENERATE_PDF:\s*(.*?)\s*\|\s*(.*?)\]', replace_pdf, text, flags=re.DOTALL)
+
+        def replace_html(match):
+            filename = match.group(1).strip()
+            content = match.group(2).strip()
+            filepath = tools.generate_html(filename, content)
+            if filepath:
+                return f"🌐 I've created your HTML file! [Download Here]({filepath})"
+            return f"🌐 HTML generation failed. Here is the content:\n\n```html\n{content}\n```"
+        text = re.sub(r'\[GENERATE_HTML:\s*(.*?)\s*\|\s*(.*?)\]', replace_html, text, flags=re.DOTALL)
+
+        return text
+
+    def deep_research(self, query, max_depth=3):
+        all_sources = []
+        all_content = []
+        current_query = query
+        for step in range(max_depth):
+            results = self.searcher.search_and_read_parallel(current_query, max_scrape=3)
+            if not results:
+                break
+            for r in results:
+                content = r.get("full_content") or r.get("snippet", "")
+                if content and len(content) > 100:
+                    all_content.append(content[:2000])
+                    all_sources.append(r.get("url", ""))
+                    facts = self.extractor.extract_facts(content, current_query)
+                    if facts and step < max_depth - 1:
+                        current_query = f"{query} {facts[0][:100]}"
+            if len(all_content) >= 6:
+                break
+        return all_content, list(set(all_sources))[:10]
+
+    def _assemble_context(self, user_query):
+        topic = self._extract_topic(user_query)
+        sources = []
+        context_snippets = []
+
+        known_facts = self.graph.recall_facts(topic, limit=5)
+        for fact, url in known_facts:
+            context_snippets.append(str(fact))
+            if url:
+                sources.append(url)
+
+        if len(context_snippets) < 3:
+            similar = self.embedder.find_similar(user_query, top_k=3)
+            for s in similar:
+                if s not in context_snippets:
+                    context_snippets.append(str(s))
+
+        needs_search = len(known_facts) < 2 and self._needs_web_search(user_query)
+        if needs_search:
+            cache_key = self.search_cache._cache_key(user_query)
+            cached = self.search_cache.get(cache_key)
+            if cached:
+                search_results = cached
+            else:
+                # Run search in a thread with a hard timeout so it never blocks the request forever
+                try:
+                    with ThreadPoolExecutor(max_workers=1) as _sex:
+                        _fut = _sex.submit(self.searcher.search_and_read_parallel, user_query, 2)
+                        search_results = _fut.result(timeout=_SEARCH_TIMEOUT_S)
+                except FuturesTimeoutError:
+                    print(f"⚠️ Search timed out after {_SEARCH_TIMEOUT_S}s — skipping web context")
+                    search_results = []
+                except Exception as _se:
+                    print(f"⚠️ Search error: {_se}")
+                    search_results = []
+                self.search_cache.set(cache_key, search_results)
+
+            for r in search_results:
+                snippet = r.get("full_content") or r.get("snippet", "")
+                url = r.get("url", "")
+                if snippet:
+                    context_snippets.append(str(snippet)[:500])
+                    if url:
+                        sources.append(url)
+            self.executor.submit(self._background_learn, search_results, topic)
+
+        return context_snippets, sources, topic
+
+    def think_stream(self, user_query, user_profile_dict=None, history=None, user_model=None, user_id=None, temperature=None, max_tokens=None, enable_thinking=None, reasoning_budget=None):
+        if self.orchestrator:
+            routed = False
+            for token, srcs in self.orchestrator.run_stream(user_query, {
+                "history": history,
+                "profile": user_profile_dict,
+                "user_model": user_model,
+                "user_id": user_id
+            }):
+                if token is None and srcs is None:
+                    break
+                routed = True
+                yield token, srcs
+            if routed:
+                self.executor.submit(self._extract_and_save_profile, user_query, user_id)
+                return
+
+        context_snippets, sources, topic = self._assemble_context(user_query)
+        profile_context = self._format_profile(user_profile_dict)
+        system_content = self._build_system(profile_context, context_snippets, user_model)
+
+        messages = [{"role": "system", "content": system_content}]
+        messages += self._format_history(history)
+        messages.append({"role": "user", "content": str(user_query)})
+
+        stream = self._call_llm(
+            messages, stream=True,
+            temperature=temperature, max_tokens=max_tokens,
+            enable_thinking=enable_thinking, reasoning_budget=reasoning_budget
+        )
+        for chunk in stream:
+            try:
+                if not getattr(chunk, 'choices', None) or len(chunk.choices) == 0:
+                    continue
+                delta = chunk.choices[0].delta
+                if delta and delta.content:
+                    yield delta.content, None
+            except (IndexError, AttributeError):
+                continue
+
+        yield "", sources
+        self.executor.submit(self._extract_and_save_profile, user_query, user_id)
+        self.graph.mark_topic_learned(topic)
+
+    def think(self, user_query, user_profile_dict=None, history=None, user_model=None, user_id=None, temperature=None, max_tokens=None, enable_thinking=None, reasoning_budget=None):
+        if self.orchestrator:
+            result, sources = self.orchestrator.run(user_query, {
+                "history": history,
+                "profile": user_profile_dict,
+                "user_model": user_model,
+                "user_id": user_id
+            })
+            if result:
+                self.executor.submit(self._extract_and_save_profile, user_query, user_id)
+                return self._process_tool_tags(result), sources
+
+        context_snippets, sources, topic = self._assemble_context(user_query)
+        profile_context = self._format_profile(user_profile_dict)
+        system_content = self._build_system(profile_context, context_snippets, user_model)
+
+        messages = [{"role": "system", "content": system_content}]
+        messages += self._format_history(history)
+        messages.append({"role": "user", "content": str(user_query)})
+
+        response = self._call_llm(
+            messages, stream=False,
+            temperature=temperature, max_tokens=max_tokens,
+            enable_thinking=enable_thinking, reasoning_budget=reasoning_budget
+        )
+        answer = response.choices[0].message.content
+        answer = self._process_tool_tags(answer)
+
+        self.executor.submit(self._extract_and_save_profile, user_query, user_id)
+        self.graph.mark_topic_learned(topic)
+        return answer, sources
+
+    def _format_profile(self, profile_dict):
+        if not profile_dict:
+            return ""
+        items = "\n".join(f"- {k}: {v}" for k, v in profile_dict.items())
+        return f"Known facts about the user:\n{items}"
+
+    def _format_history(self, history):
+        if not history:
+            return []
+        formatted = []
+        for h in history[-16:]:
+            if isinstance(h, dict):
+                role = str(h.get("role", "user"))
+                content = str(h.get("content", ""))[:500]
+                formatted.append({"role": role, "content": content})
+            elif isinstance(h, (list, tuple)) and len(h) >= 2:
+                formatted.append({"role": str(h[0]), "content": str(h[1])[:500]})
+        return formatted
+
+    def _extract_topic(self, query):
+        stopwords = {
+            "what", "is", "the", "how", "does", "who", "are", "was", "tell",
+            "me", "about", "explain", "a", "an", "can", "do", "please", "help",
+            "could", "would", "should", "give", "show", "find", "get", "make",
+        }
+        words = re.sub(r"[^\w\s]", "", query.lower()).split()
+        topic_words = [w for w in words if w not in stopwords and len(w) > 2]
+        return " ".join(topic_words[:5]) or query[:50]
+
+    def _background_learn(self, search_results, topic):
+        try:
+            for r in search_results:
+                snippet = r.get("full_content") or r.get("snippet", "")
+                url = r.get("url", "")
+                if snippet:
+                    new_facts = self.extractor.extract_facts(snippet, topic)
+                    for fact in new_facts[:10]:
+                        self.graph.store_fact(topic, fact, url)
+                        self.embedder.store(fact, {"topic": topic, "source": url})
+        except Exception as e:
+            print(f"⚠️ Background learning error: {e}")
+
+    def _extract_and_save_profile(self, query, user_id=None):
+        try:
+            from memory import ConversationMemory
+            mem = ConversationMemory()
+            for pattern, key in _PROFILE_PATTERNS:
+                match = re.search(pattern, query, re.IGNORECASE)
+                if match:
+                    value = match.group(1).strip().rstrip(".,!?")
+                    if len(value) > 1 and user_id:
+                        mem.update_user_profile(user_id, key, value.capitalize())
+        except Exception as e:
+            print(f"⚠️ Profile extraction error: {e}")

embedder.py

@@ -0,0 +1,72 @@
+# core/embedder.py
+# CRITICAL: Must set BEFORE importing chromadb to suppress telemetry
+import os
+os.environ["ANONYMIZED_TELEMETRY"] = "False"
+
+import hashlib
+import chromadb
+from chromadb.config import Settings
+
+class LocalEmbedder:
+    def __init__(self, persist_dir=None):
+        self.persist_dir = persist_dir or "/data/invicta_data/vectors"
+        os.makedirs(self.persist_dir, exist_ok=True)
+        self.client = chromadb.Client(Settings(
+            persist_directory=self.persist_dir,
+            anonymized_telemetry=False
+        ))
+        self.collection = self.client.get_or_create_collection("invicta_embeddings")
+        self._embedding_function = None
+        self._embedding_ok = None
+
+    def _ensure_embedding_function(self):
+        """Lazy-load the embedding model so startup isn't blocked by a download."""
+        if self._embedding_ok is None:
+            try:
+                from chromadb.utils import embedding_functions
+                self._embedding_function = embedding_functions.DefaultEmbeddingFunction()
+                self._embedding_ok = True
+                print("✅ Embedding function loaded")
+            except Exception as e:
+                print(f"⚠️ Embedding function unavailable: {e}")
+                self._embedding_ok = False
+        return self._embedding_ok
+
+    def _doc_id(self, text):
+        return hashlib.md5(text.strip().lower().encode()).hexdigest()
+
+    def store(self, text, metadata=None):
+        if not text or not text.strip():
+            return
+        if not self._ensure_embedding_function():
+            return
+        doc_id = self._doc_id(text)
+        meta = metadata or {}
+        existing = self.collection.get(ids=[doc_id])
+        if existing and existing.get("ids") and len(existing["ids"]) > 0:
+            return
+        self.collection.add(
+            ids=[doc_id],
+            documents=[text],
+            metadatas=[meta],
+            embeddings=[self._embedding_function([text])[0]]
+        )
+
+    def find_similar(self, query, top_k=5):
+        if not self._ensure_embedding_function():
+            return []
+        embedding = self._embedding_function([query])[0]
+        results = self.collection.query(
+            query_embeddings=[embedding],
+            n_results=top_k,
+            include=["documents", "distances"]
+        )
+        docs = []
+        if results and results.get("documents"):
+            for doc, dist in zip(results["documents"][0], results["distances"][0]):
+                if dist < 0.8:
+                    docs.append(doc)
+        return docs
+
+    def count(self):
+        return self.collection.count()

extractor.py

@@ -0,0 +1,108 @@
+# core/extractor.py
+import re
+import hashlib
+
+
+class FactExtractor:
+    # Simple sentence splitter: split on .!? followed by space and a capital letter.
+    # We skip the variable-width lookbehind which causes the re.error.
+    _SENTENCE_SPLIT = re.compile(r'(?<=[.!?])\s+(?=[A-Z])')
+
+    def extract_facts(self, text, topic=""):
+        """Extract high-quality, atomic fact sentences from raw text."""
+        if not text or not text.strip():
+            return []
+
+        # Clean text
+        text = re.sub(r"\s+", " ", text).strip()
+        text = re.sub(r"[^\x20-\x7E\n]", " ", text)  # Remove non-ASCII junk
+
+        # Split into sentences
+        sentences = self._SENTENCE_SPLIT.split(text)
+
+        topic_lower = topic.lower()
+        scored = []
+        seen_hashes = set()
+
+        for s in sentences:
+            s = s.strip()
+            if not s:
+                continue
+
+            words = s.split()
+            word_count = len(words)
+
+            # Filter by length
+            if not (8 <= word_count <= 120):
+                continue
+
+            # Skip sentences that look like navigation/boilerplate
+            if self._is_boilerplate(s):
+                continue
+
+            # Deduplicate by content hash (first 80 chars)
+            fingerprint = hashlib.md5(s[:80].lower().encode()).hexdigest()
+            if fingerprint in seen_hashes:
+                continue
+            seen_hashes.add(fingerprint)
+
+            # Score: higher if contains topic keyword
+            score = self._score_sentence(s, topic_lower, word_count)
+            scored.append((score, s))
+
+        # Sort by score (highest first), return top 20
+        scored.sort(key=lambda x: x[0], reverse=True)
+        return [s for _, s in scored[:20]]
+
+    def _score_sentence(self, sentence, topic, word_count):
+        """Score a sentence by information density and topic relevance."""
+        score = 0.0
+        s_lower = sentence.lower()
+
+        # Topic match
+        if topic and topic in s_lower:
+            score += 3.0
+
+        # Contains numbers/data (more factual)
+        if re.search(r"\d", sentence):
+            score += 1.5
+
+        # Contains proper nouns (capitalized words not at sentence start)
+        proper_nouns = re.findall(r"(?<!\. )\b[A-Z][a-z]{2,}\b", sentence[5:])
+        score += min(len(proper_nouns) * 0.5, 2.0)
+
+        # Penalize very short or very long
+        if word_count < 12:
+            score -= 0.5
+        if word_count > 80:
+            score -= 1.0
+
+        # Penalize question sentences (less factual)
+        if sentence.endswith("?"):
+            score -= 1.0
+
+        return score
+
+    def _is_boilerplate(self, sentence):
+        """Detect navigation, cookie notices, ads, etc."""
+        s_lower = sentence.lower()
+        boilerplate_phrases = [
+            "cookie", "privacy policy", "terms of service", "click here",
+            "subscribe", "newsletter", "all rights reserved", "copyright ©",
+            "sign up", "log in", "login", "register now", "follow us",
+            "share this", "read more", "learn more →", "advertisement",
+            "sponsored", "skip to content", "back to top",
+        ]
+        if any(p in s_lower for p in boilerplate_phrases):
+            return True
+
+        # All caps (likely a heading or ad)
+        if sentence.isupper() and len(sentence) > 10:
+            return True
+
+        # Very few unique words (repetitive/spammy)
+        words = sentence.lower().split()
+        if len(words) > 5 and len(set(words)) / len(words) < 0.4:
+            return True
+
+        return False

knowledge_graph.py

@@ -0,0 +1,212 @@
+# core/knowledge_graph.py
+import sqlite3
+from datetime import datetime
+import threading
+import os
+import hashlib
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+class KnowledgeGraph:
+    def __init__(self, db_path=None):
+        self.db_path = db_path or "/data/invicta_data/knowledge.db"
+        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
+        self.lock = threading.Lock()
+        self._init_tables()
+
+    def _get_cursor(self):
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        conn.execute("PRAGMA foreign_keys = ON")
+        conn.row_factory = sqlite3.Row
+        return conn, conn.cursor()
+
+    def _init_tables(self):
+        with self.lock:
+            conn, cur = self._get_cursor()
+            try:
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS facts (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    topic TEXT NOT NULL,
+                    fact TEXT NOT NULL,
+                    fact_hash TEXT,
+                    source_url TEXT DEFAULT '',
+                    confidence REAL DEFAULT 1.0,
+                    times_seen INTEGER DEFAULT 1,
+                    learned_at TEXT DEFAULT (datetime('now'))
+                );
+            """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS entities (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    name TEXT UNIQUE NOT NULL,
+                    type TEXT,
+                    description TEXT,
+                    learned_at TEXT DEFAULT (datetime('now'))
+                );
+            """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS relationships (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    entity_a TEXT NOT NULL,
+                    relation TEXT NOT NULL,
+                    entity_b TEXT NOT NULL,
+                    source_url TEXT DEFAULT '',
+                    learned_at TEXT DEFAULT (datetime('now')),
+                    UNIQUE(entity_a, relation, entity_b)
+                );
+            """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS topics_learned (
+                    topic TEXT PRIMARY KEY,
+                    times_queried INTEGER DEFAULT 1,
+                    last_updated TEXT DEFAULT (datetime('now'))
+                );
+            """)
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_facts_topic ON facts(topic);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_facts_hash ON facts(fact_hash);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_facts_learned ON facts(learned_at);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_relationships_a ON relationships(entity_a);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_relationships_b ON relationships(entity_b);")
+                conn.commit()
+            except Exception as e:
+                conn.rollback()
+                if "already exists" not in str(e).lower() and "duplicate" not in str(e).lower():
+                    raise
+            finally:
+                conn.close()
+
+    def _fact_hash(self, fact):
+        return hashlib.md5(fact.strip().lower()[:100].encode()).hexdigest()
+
+    # ── Facts ────────────────────────────────────────────────────────────────
+    def store_fact(self, topic, fact, source_url=""):
+        fh = self._fact_hash(fact)
+        with self.lock:
+            conn, cur = self._get_cursor()
+            cur.execute("SELECT id FROM facts WHERE fact_hash = ?", (fh,))
+            existing = cur.fetchone()
+
+            if existing:
+                cur.execute(
+                    "UPDATE facts SET times_seen = times_seen + 1, confidence = min(confidence + 0.1, 2.0) WHERE id = ?",
+                    (existing["id"],)
+                )
+            else:
+                cur.execute(
+                    "INSERT INTO facts (topic, fact, fact_hash, source_url, learned_at) VALUES (?, ?, ?, ?, ?)",
+                    (topic.lower(), fact, fh, source_url, datetime.now().isoformat(sep=' ', timespec='seconds'))
+                )
+            conn.commit()
+            conn.close()
+
+    def recall_facts(self, topic, limit=10):
+        conn, cur = self._get_cursor()
+        cur.execute(
+            """SELECT fact, source_url FROM facts
+               WHERE topic LIKE ?
+               ORDER BY confidence DESC, learned_at DESC
+               LIMIT ?""",
+            (f"%{topic.lower()}%", limit)
+        )
+        rows = cur.fetchall()
+        conn.close()
+        return [(r["fact"], r["source_url"]) for r in rows]
+
+    def knows_topic(self, topic):
+        conn, cur = self._get_cursor()
+        cur.execute(
+            "SELECT COUNT(*) as c FROM facts WHERE topic LIKE ?",
+            (f"%{topic.lower()}%",)
+        )
+        row = cur.fetchone()
+        conn.close()
+        return (row["c"] if row else 0) > 0
+
+    def get_fact_count(self):
+        conn, cur = self._get_cursor()
+        cur.execute("SELECT COUNT(*) as c FROM facts")
+        row = cur.fetchone()
+        conn.close()
+        return row["c"] if row else 0
+
+    def get_recent_facts(self, limit=25):
+        conn, cur = self._get_cursor()
+        cur.execute(
+            "SELECT id, topic, fact, source_url, confidence, learned_at FROM facts ORDER BY id DESC LIMIT ?",
+            (limit,)
+        )
+        rows = cur.fetchall()
+        conn.close()
+        return [dict(r) for r in rows]
+
+
+    # ── Relationships ─────────────────────────────────────────────────────────
+    def store_relationship(self, entity_a, relation, entity_b, source=""):
+        with self.lock:
+            conn, cur = self._get_cursor()
+            cur.execute(
+                """INSERT INTO relationships (entity_a, relation, entity_b, source_url, learned_at) 
+                   VALUES (?, ?, ?, ?, ?)
+                   ON CONFLICT DO NOTHING""",
+                (entity_a.lower(), relation.lower(), entity_b.lower(), source, datetime.now().isoformat(sep=' ', timespec='seconds'))
+            )
+            conn.commit()
+            conn.close()
+
+    def recall_relationships(self, entity, limit=10):
+        conn, cur = self._get_cursor()
+        cur.execute(
+            """SELECT entity_a, relation, entity_b FROM relationships
+               WHERE entity_a LIKE ? OR entity_b LIKE ?
+               LIMIT ?""",
+            (f"%{entity.lower()}%", f"%{entity.lower()}%", limit)
+        )
+        rows = cur.fetchall()
+        conn.close()
+        return [(r["entity_a"], r["relation"], r["entity_b"]) for r in rows]
+
+    # ── Topics ────────────────────────────────────────────────────────────────
+    def mark_topic_learned(self, topic):
+        with self.lock:
+            conn, cur = self._get_cursor()
+            cur.execute("""
+                INSERT INTO topics_learned (topic, times_queried, last_updated)
+                VALUES (?, 1, ?)
+                ON CONFLICT(topic) DO UPDATE SET
+                    times_queried = topics_learned.times_queried + 1,
+                    last_updated = excluded.last_updated
+            """, (topic.lower(), datetime.now().isoformat(sep=' ', timespec='seconds')))
+            conn.commit()
+            conn.close()
+
+    def get_top_topics(self, limit=10):
+        conn, cur = self._get_cursor()
+        cur.execute(
+            "SELECT topic, times_queried FROM topics_learned ORDER BY times_queried DESC LIMIT ?",
+            (limit,)
+        )
+        rows = cur.fetchall()
+        conn.close()
+        return [(r["topic"], r["times_queried"]) for r in rows]
+
+    # ── Maintenance ───────────────────────────────────────────────────────────
+    def prune_old_facts(self, max_facts=5000):
+        conn, cur = self._get_cursor()
+        cur.execute("SELECT COUNT(*) as c FROM facts")
+        count = cur.fetchone()["c"]
+        conn.close()
+        if count > max_facts:
+            with self.lock:
+                conn, cur = self._get_cursor()
+                cur.execute("""
+                    DELETE FROM facts WHERE id IN (
+                        SELECT id FROM facts
+                        ORDER BY confidence ASC, learned_at ASC
+                        LIMIT ?
+                    )
+                """, (count - max_facts,))
+                conn.commit()
+                conn.close()
+                print(f"🧹 Pruned {count - max_facts} old facts from knowledge graph")

learning.py

@@ -0,0 +1,327 @@
+# core/learning.py
+import re
+import json
+import threading
+from datetime import datetime
+from memory import ConversationMemory
+import sqlite3
+
+
+class LearningEngine:
+    """
+    Self-learning engine that continuously observes user conversations,
+    extracts communication patterns, interests, and preferences,
+    and consolidates them into an adaptive user model.
+
+    The more a user interacts, the better Invicta understands them.
+    """
+
+    def __init__(self, memory: ConversationMemory):
+        self.memory = memory
+        self._lock = threading.Lock()
+
+    # ═══════════════════════════════════════════════════════════════════════
+    # REAL-TIME MESSAGE ANALYSIS
+    # ═══════════════════════════════════════════════════════════════════════
+
+    def analyze_message(self, user_id, message):
+        if not message or not user_id:
+            return
+        try:
+            self.memory.increment_interaction_count(user_id)
+            self._detect_interests(user_id, message)
+            self._detect_communication_style(user_id, message)
+            self._detect_emotional_state(user_id, message)
+            self._detect_preferences(user_id, message)
+            self._detect_language(user_id, message)  # ← ADD THIS
+        except Exception as e:
+            print(f"⚠️ Learning analysis error: {e}")
+
+    def _detect_interests(self, user_id, message):
+        """Detect topics the user is interested in based on what they ask about."""
+        interest_patterns = [
+            # Technology
+            (r'\b(?:python|javascript|typescript|rust|golang|programming|coding|developer|software|algorithm|api|database|web\s*dev|machine\s*learning|deep\s*learning|ai|artificial\s*intelligence|neural\s*network|data\s*science|cybersecurity|blockchain|crypto|cloud\s*computing|devops)\b', 'technology'),
+            # Science
+            (r'\b(?:physics|chemistry|biology|quantum|astronomy|space|nasa|evolution|genetics|neuroscience|psychology|math|calculus|statistics|research)\b', 'science'),
+            # Business & Finance
+            (r'\b(?:startup|business|entrepreneur|invest|stock|market|finance|crypto|trading|marketing|sales|revenue|profit|startup|company|venture)\b', 'business'),
+            # Creative
+            (r'\b(?:design|art|music|writing|creative|paint|draw|compose|fiction|novel|poetry|photography|filmmaking|video|animation)\b', 'creative'),
+            # Health & Fitness
+            (r'\b(?:fitness|workout|exercise|diet|nutrition|health|medical|yoga|meditation|mental\s*health|therapy|wellness|gym|running|weight)\b', 'health'),
+            # Gaming
+            (r'\b(?:game|gaming|gamer|xbox|playstation|nintendo|steam|esports|rpg|fps|mmorpg|minecraft|fortnite)\b', 'gaming'),
+            # Travel
+            (r'\b(?:travel|trip|vacation|flight|hotel|country|city|backpack|adventure|explore|destination|tourist|visa)\b', 'travel'),
+            # Food & Cooking
+            (r'\b(?:recipe|cooking|food|cuisine|bake|chef|restaurant|meal|ingredient|kitchen|dish)\b', 'food'),
+            # Education
+            (r'\b(?:study|university|college|school|learn|course|degree|exam|academic|education|teach|student|homework|tutorial)\b', 'education'),
+            # Philosophy & Religion
+            (r'\b(?:philosophy|existential|meaning\s*of\s*life|moral|ethics|religion|spiritual|meditation|mindfulness|consciousness)\b', 'philosophy'),
+        ]
+
+        msg_lower = message.lower()
+        for pattern, category in interest_patterns:
+            matches = re.findall(pattern, msg_lower)
+            if matches:
+                # Store each matched keyword as an interest observation
+                for match in matches:
+                    keyword = match.strip().lower()
+                    self.memory.store_learning(
+                        user_id=user_id,
+                        category="interest",
+                        key=keyword,
+                        value=f"Interested in {category}: {keyword}",
+                        confidence=0.4
+                    )
+
+    def _detect_language(self, user_id, message):
+        """Detect if user is writing in a non-English language and store preference."""
+        # Simple detection for common scripts
+        import re
+        if re.search(r'[\u0900-\u097F]', message):  # Devanagari (Hindi)
+            self.memory.store_learning(user_id, "preference", "language", "hindi", 0.6)
+        elif re.search(r'[\u0600-\u06FF]', message):  # Arabic
+            self.memory.store_learning(user_id, "preference", "language", "arabic", 0.6)
+        elif re.search(r'[\u3040-\u30FF]', message):  # Japanese
+            self.memory.store_learning(user_id, "preference", "language", "japanese", 0.6)
+        elif re.search(r'[\u4E00-\u9FFF]', message):  # Chinese
+            self.memory.store_learning(user_id, "preference", "language", "chinese", 0.6)
+        elif re.search(r'[\uAC00-\uD7AF]', message):  # Korean
+            self.memory.store_learning(user_id, "preference", "language", "korean", 0.6)
+        elif re.search(r'[áéíóúñ¿¡]', message, re.IGNORECASE):  # Spanish
+            self.memory.store_learning(user_id, "preference", "language", "spanish", 0.3)
+
+    def _detect_communication_style(self, user_id, message):
+        """Detect how the user communicates — formality, length, tone."""
+        # ── FIX: msg_lower was missing here, causing the crash ──
+        msg_lower = message.lower()
+        words = message.split()
+        word_count = len(words)
+
+        # ── Formality detection ──
+        formal_indicators = sum(1 for w in words if w.lower() in {
+            "therefore", "however", "furthermore", "consequently", "nevertheless",
+            "accordingly", "moreover", "henceforth", "would", "shall", "may",
+            "perhaps", "kindly", "regards", "sincerely", "please", "respectfully",
+        })
+        casual_indicators = sum(1 for w in words if w.lower() in {
+            "lol", "haha", "yeah", "nah", "yep", "nope", "gonna", "wanna",
+            "gotta", "dunno", "kinda", "sorta", "btw", "omg", "wtf", "lmao",
+            "bruh", "dude", "hey", "yo", "sup", "ngl", "fr", "tbh", "imo",
+        })
+        contractions = sum(1 for w in words if re.match(r"\w+'\w+", w))
+
+        if formal_indicators > 2 or (formal_indicators > 0 and casual_indicators == 0 and contractions == 0):
+            self.memory.store_learning(user_id, "style", "formality_signal", "formal", 0.3)
+        elif casual_indicators > 0 or contractions > 2:
+            self.memory.store_learning(user_id, "style", "formality_signal", "casual", 0.3)
+
+        # ── Response length preference ──
+        if word_count <= 5:
+            self.memory.store_learning(user_id, "style", "length_signal", "short", 0.25)
+        elif word_count <= 20:
+            self.memory.store_learning(user_id, "style", "length_signal", "medium", 0.2)
+        else:
+            self.memory.store_learning(user_id, "style", "length_signal", "detailed", 0.25)
+
+        # ── Tone detection ──
+        # Question-heavy = curious
+        questions = message.count('?')
+        exclamations = message.count('!')
+        emojis = len(re.findall(r'[🔥💡❤️👍😊😂🤔👏🙌✨🎯💪🚀🧠⭐🌟💫]', message))
+
+        if questions >= 2:
+            self.memory.store_learning(user_id, "style", "tone_signal", "curious", 0.3)
+        if exclamations >= 2 or emojis >= 2:
+            self.memory.store_learning(user_id, "style", "tone_signal", "enthusiastic", 0.3)
+        if any(phrase in msg_lower for phrase in ['help me', 'can you', 'please', 'how do i', 'how to']):
+            self.memory.store_learning(user_id, "style", "tone_signal", "practical", 0.25)
+
+        # Detect humor/wit
+        humor_words = ['joke', 'funny', 'lol', 'haha', 'lmao', 'pun', 'humor']
+        if any(w in msg_lower for w in humor_words):
+            self.memory.store_learning(user_id, "style", "tone_signal", "witty", 0.35)
+
+    def _detect_emotional_state(self, user_id, message):
+        """Detect the user's emotional state from their messages."""
+        msg_lower = message.lower()
+
+        # Stress/frustration
+        stress_words = ['stressed', 'frustrated', 'overwhelmed', 'anxious', 'worried',
+                        'tired', 'exhausted', 'burnt out', 'burnout', 'struggling',
+                        'can\'t handle', 'too much', 'losing hope']
+        if any(w in msg_lower for w in stress_words):
+            self.memory.store_learning(user_id, "emotion", "recent_state", "stressed", 0.5)
+            self.memory.store_learning(user_id, "emotion", "stress_prone", "yes", 0.2)
+
+        # Happiness/excitement
+        happy_words = ['excited', 'happy', 'great', 'awesome', 'amazing', 'wonderful',
+                       'fantastic', 'love it', 'so good', 'thrilled', 'pumped', 'stoked']
+        if any(w in msg_lower for w in happy_words):
+            self.memory.store_learning(user_id, "emotion", "recent_state", "happy", 0.5)
+
+        # Sadness
+        sad_words = ['sad', 'depressed', 'lonely', 'miss', 'lost', 'heartbroken',
+                     'crying', 'hurt', 'pain', 'grief', 'mourn']
+        if any(w in msg_lower for w in sad_words):
+            self.memory.store_learning(user_id, "emotion", "recent_state", "sad", 0.5)
+
+        # Curiosity
+        curious_words = ['wonder', 'curious', 'interesting', 'fascinating', 'tell me more',
+                         'how come', 'why is', 'what if', 'explain']
+        if any(w in msg_lower for w in curious_words):
+            self.memory.store_learning(user_id, "emotion", "recent_state", "curious", 0.4)
+
+    def _detect_preferences(self, user_id, message):
+        """Detect explicit and implicit preferences."""
+        msg_lower = message.lower()
+
+        # Explicit preferences
+        pref_patterns = [
+            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(short|brief|concise)", "short"),
+            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(long|detailed|thorough|in.depth|comprehensive)", "detailed"),
+            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(simple|easy|basic|beginner)", "simple"),
+            (r"i (?:prefer|like|want|need) (?:my (?:answers?|responses?) )?(?:to be )?(technical|advanced|complex)", "technical"),
+            (r"(?:don't|do not) (?:use|give me) (?:code|programming)", "no_code"),
+            (r"(?:give|show) me (?:the )?code", "wants_code"),
+            (r"(?:give|show) me (?:an )?example", "wants_examples"),
+            (r"(?:step.by.step|step by step)", "step_by_step"),
+        ]
+
+        for pattern, pref in pref_patterns:
+            if re.search(pattern, msg_lower):
+                self.memory.store_learning(user_id, "preference", pref, "true", 0.6)
+
+    # ═══════════════════════════════════════════════════════════════════════
+    # CONSOLIDATION (runs periodically)
+    # ═══════════════════════════════════════════════════════════════════════
+
+    def run_daily_consolidation(self):
+        """
+        Consolidate raw learning observations into user insights.
+        Called periodically by the background thread.
+        """
+        try:
+            # Get all users who have learning data
+            all_users = self._get_active_users()
+            for user_id in all_users:
+                try:
+                    self._consolidate_user(user_id)
+                except Exception as e:
+                    print(f"⚠️ Consolidation error for user {user_id}: {e}")
+        except Exception as e:
+            print(f"⚠️ Daily consolidation error: {e}")
+
+    def _get_active_users(self):
+        """Get list of user IDs that have learning data."""
+        db_path = self.memory.db_path
+        conn = sqlite3.connect(db_path)
+        conn.row_factory = sqlite3.Row
+        cur = conn.cursor()
+        cur.execute("SELECT DISTINCT user_id FROM user_learning")
+        rows = cur.fetchall()
+        conn.close()
+        return [row["user_id"] for row in rows]
+
+    def _consolidate_user(self, user_id):
+        """Consolidate all learning data for one user into insights."""
+        insights = self.memory.get_user_insights(user_id) or {}
+
+        # ── Formality ──
+        style_data = self.memory.get_learning_by_category(user_id, "style")
+        formality_signals = []
+        for key, val in style_data.items():
+            if key == "formality_signal":
+                formality_signals.append(val)
+
+        if formality_signals:
+            # Most frequent signal wins
+            from collections import Counter
+            most_common = Counter(formality_signals).most_common(1)[0][0]
+            insights["preferred_formality"] = most_common
+
+        # ── Response length ──
+        length_signals = [v for k, v in style_data.items() if k == "length_signal"]
+        if length_signals:
+            from collections import Counter
+            most_common = Counter(length_signals).most_common(1)[0][0]
+            insights["preferred_response_length"] = most_common
+
+        # ── Tone ──
+        tone_signals = [v for k, v in style_data.items() if k == "tone_signal"]
+        if tone_signals:
+            from collections import Counter
+            most_common = Counter(tone_signals).most_common(1)[0][0]
+            insights["preferred_tone"] = most_common
+
+        # ── Topics of interest ──
+        top_topics = self.memory.get_top_learning_topics(user_id, limit=15)
+        interest_list = [t["key"] for t in top_topics if t.get("times_observed", 0) >= 2]
+        if interest_list:
+            insights["topics_of_interest"] = interest_list[:10]
+
+        # ── Communication patterns ──
+        patterns = {}
+        pref_data = self.memory.get_learning_by_category(user_id, "preference")
+        for k, v in pref_data.items():
+            patterns[k] = v
+        if patterns:
+            insights["communication_patterns"] = patterns
+
+        # Save consolidated insights
+        self.memory.update_user_insights(user_id, insights)
+
+        # Prune old learning data
+        self.memory.prune_old_learning(user_id, max_entries=500)
+
+        print(f"🧠 Consolidated learning for user {user_id}")
+
+    # ═══════════════════════════════════════════════════════════════════════
+    # USER MODEL (used by Brain to adapt responses)
+    # ═══════════════════════════════════════════════════════════════════════
+
+    def get_user_model(self, user_id):
+        """
+        Build the current user model from consolidated insights
+        + real-time learning data. Used by Brain._build_adaptation_prompt().
+        """
+        if not user_id:
+            return None
+
+        insights = self.memory.get_user_insights(user_id)
+        model = {}
+
+        # Formality
+        model["formality"] = insights.get("preferred_formality", "casual") if insights else "casual"
+
+        # Response length
+        model["response_length"] = insights.get("preferred_response_length", "medium") if insights else "medium"
+
+        # Tone
+        model["tone"] = insights.get("preferred_tone", "friendly") if insights else "friendly"
+
+        # Interests
+        model["interests"] = insights.get("topics_of_interest", []) if insights else []
+
+        # Preferences
+        if insights and insights.get("communication_patterns"):
+            model["preferences"] = insights["communication_patterns"]
+        else:
+            model["preferences"] = {}
+
+        # Emotional state (from real-time learning)
+        emotion_data = self.memory.get_learning_by_category(user_id, "emotion")
+        recent_state = emotion_data.get("recent_state", "")
+        if recent_state:
+            model["recent_emotional_state"] = recent_state
+
+        # Interaction count
+        if insights:
+            model["interaction_count"] = insights.get("total_interactions", 0)
+        else:
+            model["interaction_count"] = 0
+
+        return model

memory.py

@@ -0,0 +1,1037 @@
+# core/memory.py
+import sqlite3
+from datetime import datetime
+import threading
+import os
+import json
+import secrets
+import hashlib
+import time
+import re
+from werkzeug.security import generate_password_hash, check_password_hash
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+class ConversationMemory:
+    def __init__(self, db_path=None):
+        self.db_path = db_path or "/data/invicta_data/memory.db"
+        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
+        self.lock = threading.Lock()
+        self._init_tables()
+
+    def _get_connection(self):
+        """Create a new SQLite connection with WAL mode and busy timeout."""
+        conn = sqlite3.connect(
+            self.db_path,
+            check_same_thread=False,
+            timeout=30.0,
+            isolation_level=None
+        )
+        conn.execute("PRAGMA foreign_keys = ON")
+        conn.execute("PRAGMA journal_mode = WAL")
+        conn.execute("PRAGMA synchronous = NORMAL")
+        conn.execute("PRAGMA temp_store = MEMORY")
+        conn.execute("PRAGMA mmap_size = 30000000000")
+        conn.execute("PRAGMA busy_timeout = 30000")
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _get_cursor(self):
+        """Legacy compatibility wrapper."""
+        conn = self._get_connection()
+        return conn, conn.cursor()
+
+    def _execute_with_retry(self, sql, params=(), max_retries=5, sleep_base=0.05):
+        """Execute SQL with automatic retry on database lock."""
+        for attempt in range(max_retries):
+            conn = None
+            try:
+                conn = self._get_connection()
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(sql, params)
+                conn.commit()
+                return cur
+            except sqlite3.OperationalError as e:
+                if conn:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                if "database is locked" in str(e).lower() and attempt < max_retries - 1:
+                    time.sleep(sleep_base * (attempt + 1))
+                    continue
+                raise
+            finally:
+                if conn:
+                    try:
+                        conn.close()
+                    except Exception:
+                        pass
+        return None
+
+    def _init_tables(self):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS users (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    username TEXT UNIQUE NOT NULL,
+                    password_hash TEXT NOT NULL,
+                    google_id TEXT,
+                    email TEXT,
+                    created_at TEXT DEFAULT (datetime('now'))
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS auth_tokens (
+                    token TEXT PRIMARY KEY,
+                    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                    created_at TEXT DEFAULT (datetime('now')),
+                    last_used TEXT DEFAULT (datetime('now'))
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS chat_sessions (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                    title TEXT,
+                    created_at TEXT DEFAULT (datetime('now'))
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS conversations (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
+                    session_id INTEGER REFERENCES chat_sessions(id) ON DELETE CASCADE,
+                    role TEXT NOT NULL,
+                    content TEXT NOT NULL,
+                    timestamp TEXT DEFAULT (datetime('now'))
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS user_profile (
+                    user_id INTEGER NOT NULL DEFAULT 0 REFERENCES users(id) ON DELETE CASCADE,
+                    key TEXT NOT NULL,
+                    value TEXT,
+                    category TEXT DEFAULT 'general',
+                    updated_at TEXT DEFAULT (datetime('now')),
+                    PRIMARY KEY (user_id, key)
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS user_learning (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                    category TEXT NOT NULL,
+                    key TEXT NOT NULL,
+                    value TEXT NOT NULL,
+                    confidence REAL DEFAULT 0.5,
+                    times_observed INTEGER DEFAULT 1,
+                    first_seen TEXT DEFAULT (datetime('now')),
+                    updated_at TEXT DEFAULT (datetime('now')),
+                    UNIQUE(user_id, category, key)
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS user_insights (
+                    user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+                    preferred_formality TEXT DEFAULT 'casual',
+                    preferred_response_length TEXT DEFAULT 'medium',
+                    preferred_tone TEXT DEFAULT 'friendly',
+                    topics_of_interest TEXT DEFAULT '[]',
+                    communication_patterns TEXT DEFAULT '{}',
+                    total_interactions INTEGER DEFAULT 0,
+                    last_analyzed TEXT DEFAULT (datetime('now')),
+                    updated_at TEXT DEFAULT (datetime('now'))
+                );
+                """)
+                cur.execute("""
+                CREATE TABLE IF NOT EXISTS api_keys (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                    key_hash TEXT NOT NULL,
+                    name TEXT DEFAULT 'API Key',
+                    prefix TEXT DEFAULT '',
+                    created_at TEXT DEFAULT (datetime('now')),
+                    last_used TEXT,
+                    revoked INTEGER DEFAULT 0,
+                    rate_limit INTEGER DEFAULT 60
+                );
+                """)
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_conversations_user ON conversations(user_id);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_conversations_session ON conversations(session_id);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_conversations_user_session ON conversations(user_id, session_id);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_conversations_timestamp ON conversations(timestamp);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_learning_user ON user_learning(user_id);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_learning_category ON user_learning(user_id, category);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_learning_unique ON user_learning(user_id, category, key);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_auth_tokens_user ON auth_tokens(user_id);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_api_keys_user ON api_keys(user_id);")
+                cur.execute("CREATE INDEX IF NOT EXISTS idx_api_keys_hash ON api_keys(key_hash);")
+                conn.commit()
+            except Exception as e:
+                conn.rollback()
+                if "already exists" not in str(e).lower() and "duplicate" not in str(e).lower():
+                    raise
+            finally:
+                conn.close()
+
+    # ── Auth Tokens ────────────────────────────────────────────────────────
+    def create_token(self, user_id):
+        token = secrets.token_urlsafe(32)
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "INSERT INTO auth_tokens (token, user_id, created_at, last_used) VALUES (?, ?, ?, ?)",
+                    (token, user_id, datetime.now().isoformat(sep=' ', timespec='seconds'), datetime.now().isoformat(sep=' ', timespec='seconds'))
+                )
+                conn.commit()
+            finally:
+                conn.close()
+        return token
+
+    def validate_token(self, token):
+        if not token or len(token) < 10:
+            return None
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute("SELECT user_id FROM auth_tokens WHERE token = ?", (token,))
+            row = cur.fetchone()
+            return row["user_id"] if row else None
+        finally:
+            conn.close()
+
+    def touch_token(self, token):
+        """Update token last_used with retry logic for concurrent access."""
+        now = datetime.now().isoformat(sep=' ', timespec='seconds')
+        for attempt in range(5):
+            conn = None
+            try:
+                conn = self._get_connection()
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "UPDATE auth_tokens SET last_used = ? WHERE token = ?",
+                    (now, token)
+                )
+                conn.commit()
+                return
+            except sqlite3.OperationalError as e:
+                if conn:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                if "database is locked" in str(e).lower() and attempt < 4:
+                    time.sleep(0.05 * (attempt + 1))
+                    continue
+                raise
+            finally:
+                if conn:
+                    try:
+                        conn.close()
+                    except Exception:
+                        pass
+
+    def delete_token(self, token):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("DELETE FROM auth_tokens WHERE token = ?", (token,))
+                conn.commit()
+            finally:
+                conn.close()
+
+    def revoke_user_tokens(self, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("DELETE FROM auth_tokens WHERE user_id = ?", (user_id,))
+                conn.commit()
+            finally:
+                conn.close()
+
+    def cleanup_old_tokens(self, max_age_days=30):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "DELETE FROM auth_tokens WHERE last_used < datetime('now', '-{} days')".format(max_age_days)
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+    # ── User Auth ─────────────────────────────────────────────────────────
+    def create_user(self, username, password):
+        """Create a user with validation. Returns True on success, False on failure."""
+        # Defense-in-depth validations (also enforced in app.py)
+        if not username or not password:
+            return False
+        if len(username) < 3 or len(username) > 30:
+            return False
+        if not re.match(r'^[a-zA-Z0-9_\-]+$', username):
+            return False
+        if username.lower().startswith("__guest_"):
+            return False
+        if len(password) < 8:
+            return False
+        if not re.search(r'[A-Z]', password) or not re.search(r'[a-z]', password) or not re.search(r'[0-9]', password):
+            return False
+
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                pw_hash = generate_password_hash(password)
+                # Store username lowercase for case-normalization
+                cur.execute(
+                    "INSERT INTO users (username, password_hash) VALUES (?, ?)",
+                    (username.strip().lower(), pw_hash)
+                )
+                conn.commit()
+                return True
+            except sqlite3.IntegrityError:
+                conn.rollback()
+                return False
+            finally:
+                conn.close()
+
+    def verify_user(self, username, password):
+        if not username or not password:
+            return None
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            # Case-normalized lookup
+            cur.execute(
+                "SELECT id, password_hash FROM users WHERE username = ?", (username.strip().lower(),)
+            )
+            row = cur.fetchone()
+            if row and check_password_hash(row["password_hash"], password):
+                return row["id"]
+            return None
+        finally:
+            conn.close()
+
+    def get_username(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute("SELECT username FROM users WHERE id = ?", (user_id,))
+            row = cur.fetchone()
+            return row["username"] if row else None
+        finally:
+            conn.close()
+
+    def is_guest_user(self, user_id):
+        username = self.get_username(user_id)
+        return username.startswith("__guest_") if username else True
+
+    def create_guest_user(self, username=None, password=None):
+        if not username:
+            username = f"__guest_{secrets.token_hex(8)}"
+        if not password:
+            password = secrets.token_hex(32)
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                pw_hash = generate_password_hash(password)
+                cur.execute(
+                    "INSERT INTO users (username, password_hash) VALUES (?, ?)",
+                    (username, pw_hash)
+                )
+                user_id = cur.lastrowid
+                conn.commit()
+                return user_id
+            except sqlite3.IntegrityError:
+                conn.rollback()
+                return None
+            finally:
+                conn.close()
+
+    def find_or_create_google_user(self, google_id, name, email):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("SELECT id FROM users WHERE google_id = ?", (google_id,))
+                row = cur.fetchone()
+                if row:
+                    conn.commit()
+                    return row["id"]
+
+                cur.execute("SELECT id FROM users WHERE email = ?", (email,))
+                row = cur.fetchone()
+                if row:
+                    cur.execute("UPDATE users SET google_id = ? WHERE id = ?", (google_id, row["id"]))
+                    conn.commit()
+                    return row["id"]
+
+                base_username = name.replace(" ", "_").lower()
+                base_username = ''.join(c for c in base_username if c.isalnum() or c == '_')
+                if not base_username:
+                    base_username = "user"
+
+                username = base_username
+                counter = 1
+                while True:
+                    cur.execute("SELECT id FROM users WHERE username = ?", (username,))
+                    if not cur.fetchone():
+                        break
+                    username = f"{base_username}_{counter}"
+                    counter += 1
+
+                random_pw = secrets.token_hex(32)
+                pw_hash = generate_password_hash(random_pw)
+                cur.execute(
+                    "INSERT INTO users (username, password_hash, google_id, email) VALUES (?, ?, ?, ?)",
+                    (username, pw_hash, google_id, email)
+                )
+                user_id = cur.lastrowid
+                conn.commit()
+                return user_id
+            except sqlite3.IntegrityError:
+                conn.rollback()
+                return None
+            finally:
+                conn.close()
+
+    def transfer_guest_data(self, guest_user_id, real_user_id):
+        """
+        Transfer guest user data to real user account.
+        Handles UNIQUE constraint conflicts by merging learning data.
+        """
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+
+                # 1. Transfer conversations
+                cur.execute(
+                    "UPDATE conversations SET user_id = ? WHERE user_id = ?",
+                    (real_user_id, guest_user_id)
+                )
+
+                # 2. Transfer chat sessions
+                cur.execute(
+                    "UPDATE chat_sessions SET user_id = ? WHERE user_id = ?",
+                    (real_user_id, guest_user_id)
+                )
+
+                # 3. Transfer user_profile (ON CONFLICT handles duplicates)
+                cur.execute(
+                    "SELECT user_id, key, value, category, updated_at FROM user_profile WHERE user_id = ?",
+                    (guest_user_id,)
+                )
+                for row in cur.fetchall():
+                    cur.execute(
+                        """INSERT INTO user_profile (user_id, key, value, category, updated_at)
+                           VALUES (?, ?, ?, ?, ?)
+                           ON CONFLICT (user_id, key) DO UPDATE SET
+                           value = excluded.value,
+                           category = excluded.category,
+                           updated_at = excluded.updated_at""",
+                        (real_user_id, row["key"], row["value"], row["category"], row["updated_at"])
+                    )
+                cur.execute("DELETE FROM user_profile WHERE user_id = ?", (guest_user_id,))
+
+                # 4. Transfer user_learning with conflict resolution
+                cur.execute(
+                    "SELECT category, key, value, confidence, times_observed, first_seen, updated_at "
+                    "FROM user_learning WHERE user_id = ?",
+                    (guest_user_id,)
+                )
+                guest_rows = cur.fetchall()
+
+                for row in guest_rows:
+                    cur.execute(
+                        "SELECT id, confidence, times_observed, value FROM user_learning "
+                        "WHERE user_id = ? AND category = ? AND key = ?",
+                        (real_user_id, row["category"], row["key"])
+                    )
+                    existing = cur.fetchone()
+
+                    if existing:
+                        new_confidence = max(float(existing["confidence"]), float(row["confidence"]))
+                        new_times = int(existing["times_observed"]) + int(row["times_observed"])
+                        new_value = row["value"] if float(row["confidence"]) >= float(existing["confidence"]) else existing["value"]
+                        cur.execute(
+                            "UPDATE user_learning SET value = ?, confidence = ?, times_observed = ?, updated_at = ? "
+                            "WHERE id = ?",
+                            (new_value, new_confidence, new_times,
+                             datetime.now().isoformat(sep=' ', timespec='seconds'), existing["id"])
+                        )
+                    else:
+                        cur.execute(
+                            "INSERT INTO user_learning (user_id, category, key, value, confidence, "
+                            "times_observed, first_seen, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                            (real_user_id, row["category"], row["key"], row["value"],
+                             row["confidence"], row["times_observed"], row["first_seen"],
+                             datetime.now().isoformat(sep=' ', timespec='seconds'))
+                        )
+
+                cur.execute("DELETE FROM user_learning WHERE user_id = ?", (guest_user_id,))
+
+                # 5. Transfer user_insights (ON CONFLICT handles duplicates)
+                cur.execute(
+                    "SELECT * FROM user_insights WHERE user_id = ?",
+                    (guest_user_id,)
+                )
+                guest_insights = cur.fetchone()
+                if guest_insights:
+                    insights_dict = dict(guest_insights)
+                    insights_dict.pop("user_id", None)
+                    cols = [k for k in insights_dict.keys()]
+                    vals = [insights_dict[k] for k in cols]
+                    placeholders = ",".join(["?"] * len(vals))
+                    updates = ",".join([f"{c}=excluded.{c}" for c in cols])
+                    cur.execute(
+                        f"""INSERT INTO user_insights (user_id, {','.join(cols)})
+                            VALUES (? , {placeholders})
+                            ON CONFLICT (user_id) DO UPDATE SET {updates}""",
+                        [real_user_id] + vals
+                    )
+                    cur.execute("DELETE FROM user_insights WHERE user_id = ?", (guest_user_id,))
+
+                # 6. Transfer API keys
+                cur.execute(
+                    "UPDATE api_keys SET user_id = ? WHERE user_id = ?",
+                    (real_user_id, guest_user_id)
+                )
+
+                # 7. Delete guest auth tokens
+                cur.execute("DELETE FROM auth_tokens WHERE user_id = ?", (guest_user_id,))
+
+                # 8. Delete guest user
+                cur.execute("DELETE FROM users WHERE id = ?", (guest_user_id,))
+
+                conn.commit()
+                print(f"✅ Transferred guest {guest_user_id} → user {real_user_id}")
+            except Exception as e:
+                conn.rollback()
+                print(f"❌ Transfer error: {e}")
+                raise
+            finally:
+                conn.close()
+
+    def cleanup_guest_users(self, max_age_days=14):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "SELECT id FROM users WHERE username LIKE '__guest_%' AND created_at < datetime('now', '-{} days')".format(max_age_days)
+                )
+                guest_ids = [row["id"] for row in cur.fetchall()]
+                if not guest_ids:
+                    conn.commit()
+                    return
+                placeholders = ','.join('?' * len(guest_ids))
+                cur.execute(f"DELETE FROM conversations WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM chat_sessions WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM user_learning WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM user_insights WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM auth_tokens WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM api_keys WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM user_profile WHERE user_id IN ({placeholders})", guest_ids)
+                cur.execute(f"DELETE FROM users WHERE id IN ({placeholders})", guest_ids)
+                conn.commit()
+                print(f"🧹 Cleaned up {len(guest_ids)} old guest users")
+            finally:
+                conn.close()
+
+    # ── Sessions ─────────────────────────────────────────────────────────
+    def create_session(self, user_id, title=None):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "INSERT INTO chat_sessions (user_id, title) VALUES (?, ?)",
+                    (user_id, title or "New Chat")
+                )
+                session_id = cur.lastrowid
+                conn.commit()
+                return session_id
+            finally:
+                conn.close()
+
+    def rename_session(self, session_id, title):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("UPDATE chat_sessions SET title = ? WHERE id = ?", (title[:60], session_id))
+                conn.commit()
+            finally:
+                conn.close()
+
+    def rename_session_if_default(self, session_id, title, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "SELECT title FROM chat_sessions WHERE id = ? AND user_id = ?",
+                    (session_id, user_id)
+                )
+                row = cur.fetchone()
+                if row and (row["title"] == "New Chat" or not row["title"]):
+                    cur.execute("UPDATE chat_sessions SET title = ? WHERE id = ?", (title[:60], session_id))
+                    conn.commit()
+                else:
+                    conn.commit()
+            finally:
+                conn.close()
+
+    def delete_session(self, session_id, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("DELETE FROM conversations WHERE session_id = ? AND user_id = ?", (session_id, user_id))
+                cur.execute("DELETE FROM chat_sessions WHERE id = ? AND user_id = ?", (session_id, user_id))
+                conn.commit()
+            finally:
+                conn.close()
+
+    def get_sessions(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                """SELECT cs.id, cs.title, cs.created_at,
+                          COUNT(c.id) as msg_count
+                   FROM chat_sessions cs
+                   LEFT JOIN conversations c ON c.session_id = cs.id
+                   WHERE cs.user_id = ?
+                   GROUP BY cs.id
+                   ORDER BY cs.created_at DESC
+                   LIMIT 50""",
+                (user_id,)
+            )
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    # ── Chat Messages ────────────────────────────────────────────────────
+    def save(self, role, content, user_id=None, session_id=None):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "INSERT INTO conversations (user_id, session_id, role, content, timestamp) VALUES (?, ?, ?, ?, ?)",
+                    (user_id, session_id, role, content, datetime.now().isoformat(sep=' ', timespec='seconds'))
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+    def get_recent(self, user_id, limit=20, session_id=None):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            if session_id:
+                cur.execute(
+                    "SELECT role, content, timestamp FROM conversations "
+                    "WHERE user_id = ? AND session_id = ? ORDER BY id DESC LIMIT ?",
+                    (user_id, session_id, limit)
+                )
+            else:
+                cur.execute(
+                    "SELECT role, content, timestamp FROM conversations "
+                    "WHERE user_id = ? ORDER BY id DESC LIMIT ?",
+                    (user_id, limit)
+                )
+            rows = cur.fetchall()
+            return list(reversed([(r["role"], r["content"], r["timestamp"]) for r in rows]))
+        finally:
+            conn.close()
+
+    def get_session_messages(self, session_id, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT role, content, timestamp FROM conversations "
+                "WHERE session_id = ? AND user_id = ? ORDER BY id ASC",
+                (session_id, user_id)
+            )
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def get_chat_history_list(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT id, content FROM conversations "
+                "WHERE user_id = ? AND role = 'user' ORDER BY id DESC LIMIT 20",
+                (user_id,)
+            )
+            rows = cur.fetchall()
+            return [(r["id"], r["content"]) for r in rows]
+        finally:
+            conn.close()
+
+    def search_history(self, user_id, query, limit=10):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT role, content, timestamp FROM conversations "
+                "WHERE user_id = ? AND content LIKE ? ORDER BY id DESC LIMIT ?",
+                (user_id, f"%{query}%", limit)
+            )
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def get_user_message_count(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT COUNT(*) as c FROM conversations WHERE user_id = ? AND role = 'user'",
+                (user_id,)
+            )
+            row = cur.fetchone()
+            return row["c"] if row else 0
+        finally:
+            conn.close()
+
+    # ── User Profile (user-specific) ─────────────────────────────────────
+    def update_user_profile(self, user_id, key, value, category="general"):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("""
+                    INSERT INTO user_profile (user_id, key, value, category, updated_at)
+                    VALUES (?, ?, ?, ?, ?)
+                    ON CONFLICT (user_id, key) DO UPDATE SET
+                        value = excluded.value, category = excluded.category, updated_at = excluded.updated_at
+                """, (user_id, key, str(value), category, datetime.now().isoformat(sep=' ', timespec='seconds')))
+                conn.commit()
+            finally:
+                conn.close()
+
+    def get_user_profile(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT key, value FROM user_profile WHERE user_id = ? ORDER BY key",
+                (user_id,)
+            )
+            rows = cur.fetchall()
+            return {r["key"]: r["value"] for r in rows}
+        finally:
+            conn.close()
+
+    def delete_profile_key(self, user_id, key):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("DELETE FROM user_profile WHERE user_id = ? AND key = ?", (user_id, key))
+                conn.commit()
+            finally:
+                conn.close()
+
+    def clear_profile(self, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("DELETE FROM user_profile WHERE user_id = ?", (user_id,))
+                conn.commit()
+            finally:
+                conn.close()
+
+    # ── Self-Learning ──────────────────────────────────────────────────────
+    def store_learning(self, user_id, category, key, value, confidence=0.5):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "SELECT id, confidence, times_observed FROM user_learning "
+                    "WHERE user_id = ? AND category = ? AND key = ?",
+                    (user_id, category, key)
+                )
+                existing = cur.fetchone()
+                if existing:
+                    new_confidence = min(existing["confidence"] + 0.1, 2.0)
+                    new_times = existing["times_observed"] + 1
+                    cur.execute(
+                        "UPDATE user_learning SET confidence = ?, times_observed = ?, value = ?, updated_at = ? WHERE id = ?",
+                        (new_confidence, new_times, value, datetime.now().isoformat(sep=' ', timespec='seconds'), existing["id"])
+                    )
+                else:
+                    cur.execute(
+                        "INSERT INTO user_learning (user_id, category, key, value, confidence, first_seen, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                        (user_id, category, key, value, confidence, datetime.now().isoformat(sep=' ', timespec='seconds'), datetime.now().isoformat(sep=' ', timespec='seconds'))
+                    )
+                conn.commit()
+            finally:
+                conn.close()
+
+    def get_user_learning(self, user_id, category=None):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            if category:
+                cur.execute(
+                    "SELECT category, key, value, confidence, times_observed FROM user_learning WHERE user_id = ? AND category = ? ORDER BY confidence DESC",
+                    (user_id, category)
+                )
+            else:
+                cur.execute(
+                    "SELECT category, key, value, confidence, times_observed FROM user_learning WHERE user_id = ? ORDER BY confidence DESC",
+                    (user_id,)
+                )
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def get_learning_by_category(self, user_id, category):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT key, value FROM user_learning WHERE user_id = ? AND category = ? ORDER BY confidence DESC",
+                (user_id, category)
+            )
+            rows = cur.fetchall()
+            return {r["key"]: r["value"] for r in rows}
+        finally:
+            conn.close()
+
+    def get_top_learning_topics(self, user_id, limit=10):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT key, value, confidence, times_observed FROM user_learning "
+                "WHERE user_id = ? AND category = 'interest' ORDER BY times_observed DESC, confidence DESC LIMIT ?",
+                (user_id, limit)
+            )
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def prune_old_learning(self, user_id, max_entries=500):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("SELECT COUNT(*) as c FROM user_learning WHERE user_id = ?", (user_id,))
+                count = cur.fetchone()["c"]
+                if count > max_entries:
+                    cur.execute(
+                        "DELETE FROM user_learning WHERE id IN (SELECT id FROM user_learning WHERE user_id = ? ORDER BY confidence ASC, times_observed ASC LIMIT ?)",
+                        (user_id, count - max_entries)
+                    )
+                    conn.commit()
+                else:
+                    conn.commit()
+            finally:
+                conn.close()
+
+    # ── User Insights ────────────────────────────────────────────────────
+    def update_user_insights(self, user_id, insights_dict):
+        with self.lock:
+            allowed_keys = ["preferred_formality", "preferred_response_length", "preferred_tone", "topics_of_interest", "communication_patterns", "total_interactions"]
+            insert_cols = [k for k in insights_dict if k in allowed_keys]
+            if not insert_cols:
+                return
+
+            insert_vals = [json.dumps(insights_dict[k]) if isinstance(insights_dict[k], (dict, list)) else str(insights_dict[k]) for k in insert_cols]
+            insert_cols.append("updated_at")
+            insert_vals.append(datetime.now().isoformat(sep=' ', timespec='seconds'))
+            updates = [f"{column} = excluded.{column}" for column in insert_cols]
+
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                sql = f"""INSERT INTO user_insights (user_id, {', '.join(insert_cols)}) 
+                        VALUES (?, {', '.join(['?'] * len(insert_vals))}) 
+                        ON CONFLICT (user_id) DO UPDATE SET {', '.join(updates)}"""
+                cur.execute(sql, [user_id] + insert_vals)
+                conn.commit()
+            finally:
+                conn.close()
+
+    def get_user_insights(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute("SELECT * FROM user_insights WHERE user_id = ?", (user_id,))
+            row = cur.fetchone()
+            if row:
+                result = dict(row)
+                for field in ["topics_of_interest", "communication_patterns"]:
+                    if result.get(field):
+                        try:
+                            result[field] = json.loads(result[field])
+                        except (json.JSONDecodeError, TypeError):
+                            result[field] = [] if field == "topics_of_interest" else {}
+                return result
+            return None
+        finally:
+            conn.close()
+
+    def increment_interaction_count(self, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    """INSERT INTO user_insights (user_id, total_interactions, updated_at) 
+                       VALUES (?, 1, ?) 
+                       ON CONFLICT (user_id) DO UPDATE SET 
+                       total_interactions = total_interactions + 1, updated_at = excluded.updated_at""",
+                    (user_id, datetime.now().isoformat(sep=' ', timespec='seconds'))
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+    # ── API Keys ───────────────────────────────────────────────────────────
+    def create_api_key(self, user_id, name="API Key"):
+        raw_key = "invicta_sk_" + secrets.token_urlsafe(32)
+        key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
+        prefix = raw_key[:14] + "..."
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "INSERT INTO api_keys (user_id, key_hash, name, prefix, created_at) VALUES (?, ?, ?, ?, ?)",
+                    (user_id, key_hash, name, prefix, datetime.now().isoformat(sep=' ', timespec='seconds'))
+                )
+                key_id = cur.lastrowid
+                conn.commit()
+                return key_id, raw_key
+            finally:
+                conn.close()
+
+    def validate_api_key(self, raw_key):
+        if not raw_key or not raw_key.startswith("invicta_sk_"):
+            return None
+        key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT id, user_id, revoked FROM api_keys WHERE key_hash = ?",
+                (key_hash,)
+            )
+            row = cur.fetchone()
+            if row and not row["revoked"]:
+                cur.execute(
+                    "UPDATE api_keys SET last_used = ? WHERE id = ?",
+                    (datetime.now().isoformat(sep=' ', timespec='seconds'), row["id"])
+                )
+                conn.commit()
+                return row["user_id"]
+            return None
+        finally:
+            conn.close()
+
+    def get_api_keys(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT id, name, prefix, created_at, last_used, revoked, rate_limit FROM api_keys WHERE user_id = ? ORDER BY created_at DESC",
+                (user_id,)
+            )
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def revoke_api_key(self, key_id, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute(
+                    "UPDATE api_keys SET revoked = 1 WHERE id = ? AND user_id = ?",
+                    (key_id, user_id)
+                )
+                conn.commit()
+                return cur.rowcount > 0
+            finally:
+                conn.close()
+
+    def delete_api_key(self, key_id, user_id):
+        with self.lock:
+            conn = self._get_connection()
+            try:
+                cur = conn.cursor()
+                cur.execute("BEGIN IMMEDIATE")
+                cur.execute("DELETE FROM api_keys WHERE id = ? AND user_id = ?", (key_id, user_id))
+                conn.commit()
+            finally:
+                conn.close()
+
+    # ── Stats ──────────────────────────────────────────────────────────────
+    def get_stats(self, user_id):
+        conn = self._get_connection()
+        try:
+            cur = conn.cursor()
+            cur.execute("SELECT COUNT(*) as c FROM conversations WHERE user_id = ?", (user_id,))
+            total = cur.fetchone()["c"]
+            cur.execute("SELECT COUNT(*) as c FROM chat_sessions WHERE user_id = ?", (user_id,))
+            sessions = cur.fetchone()["c"]
+            cur.execute("SELECT COUNT(*) as c FROM user_learning WHERE user_id = ?", (user_id,))
+            learning = cur.fetchone()["c"]
+            return {"total_messages": total, "total_sessions": sessions, "learning_observations": learning}
+        finally:
+            conn.close()

searcher.py

@@ -0,0 +1,168 @@
+# core/searcher.py
+import re
+import time
+import hashlib
+import threading
+import requests
+from bs4 import BeautifulSoup
+from ddgs import DDGS
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+_USER_AGENTS = [
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+]
+_ua_index = 0
+_ua_lock = threading.Lock()
+
+
+def _next_ua():
+    global _ua_index
+    with _ua_lock:
+        ua = _USER_AGENTS[_ua_index % len(_USER_AGENTS)]
+        _ua_index += 1
+    return ua
+
+
+class WebSearcher:
+    def __init__(self):
+        self.ddgs = DDGS()
+        self.scraper_pool = ThreadPoolExecutor(max_workers=3)
+        self._cache = {}
+        self._cache_ttl = 300  # 5 minutes
+        self._cache_lock = threading.Lock()
+
+    # ── Cache helpers ────────────────────────────────────────────────────────
+    def _cache_key(self, query):
+        return hashlib.md5(query.lower().strip().encode()).hexdigest()
+
+    def _cache_get(self, key):
+        with self._cache_lock:
+            entry = self._cache.get(key)
+            if entry and time.time() - entry["ts"] < self._cache_ttl:
+                return entry["value"]
+        return None
+
+    def _cache_set(self, key, value):
+        with self._cache_lock:
+            self._cache[key] = {"value": value, "ts": time.time()}
+            # Evict old entries if cache is too large
+            if len(self._cache) > 100:
+                oldest = min(self._cache.items(), key=lambda x: x[1]["ts"])
+                del self._cache[oldest[0]]
+
+    # ── Search ───────────────────────────────────────────────────────────────
+    def search(self, query, max_results=6):
+        cache_key = self._cache_key(f"search:{query}")
+        cached = self._cache_get(cache_key)
+        if cached:
+            return cached
+
+        results = []
+        seen_urls = set()
+        try:
+            for r in self.ddgs.text(query, max_results=max_results + 2):
+                url = r.get("href", "")
+                if url in seen_urls:
+                    continue
+                # Skip low-quality sources
+                if any(skip in url for skip in ["reddit.com/r/", "quora.com", "pinterest.com"]):
+                    continue
+                seen_urls.add(url)
+                results.append({
+                    "title": r.get("title", ""),
+                    "url": url,
+                    "snippet": r.get("body", ""),
+                })
+                if len(results) >= max_results:
+                    break
+        except Exception as e:
+            print(f"⚠️ Search error: {e}")
+
+        self._cache_set(cache_key, results)
+        return results
+
+    # ── Scrape ───────────────────────────────────────────────────────────────
+    def scrape_page(self, url):
+        try:
+            headers = {
+                "User-Agent": _next_ua(),
+                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+                "Accept-Language": "en-US,en;q=0.5",
+            }
+            res = requests.get(url, headers=headers, timeout=6, allow_redirects=True)
+            res.raise_for_status()
+
+            soup = BeautifulSoup(res.text, "html.parser")
+
+            # Remove boilerplate
+            for tag in soup(["script", "style", "nav", "footer", "header",
+                              "aside", "form", "noscript", "iframe", "ads",
+                              "[class*='cookie']", "[class*='banner']",
+                              "[id*='sidebar']", "[id*='nav']"]):
+                tag.decompose()
+
+            # Try to get the main content area
+            main = (
+                soup.find("main") or
+                soup.find("article") or
+                soup.find(id="content") or
+                soup.find(class_="content") or
+                soup.find(class_="post-content") or
+                soup.body
+            )
+
+            if main:
+                text = main.get_text(separator=" ", strip=True)
+            else:
+                text = soup.get_text(separator=" ", strip=True)
+
+            # Clean up whitespace and repeated chars
+            text = re.sub(r"\s{3,}", "  ", text)
+            text = re.sub(r"\.{4,}", "...", text)
+            return text[:5000]
+
+        except requests.exceptions.Timeout:
+            return None
+        except requests.exceptions.HTTPError:
+            return None
+        except Exception as e:
+            print(f"⚠️ Scrape error ({url[:40]}): {e}")
+            return None
+
+    # ── Parallel search + scrape ─────────────────────────────────────────────
+    def search_and_read_parallel(self, query, max_scrape=3, max_results=8):
+        """Enhanced search with deeper content extraction."""
+        results = self.search(query, max_results=max_results)
+        if not results:
+            return []
+
+        future_to_idx = {}
+        for i, r in enumerate(results[:max_scrape]):
+            future = self.scraper_pool.submit(self.scrape_page, r["url"])
+            future_to_idx[future] = i
+
+        for future in as_completed(future_to_idx, timeout=15):
+            idx = future_to_idx[future]
+            try:
+                content = future.result()
+                if content:
+                    results[idx]["full_content"] = content
+                    # Extract key facts
+                    sentences = content.split('.')[:20]
+                    results[idx]["key_points"] = [s.strip() for s in sentences if len(s) > 40 and len(s) < 200]
+            except Exception:
+                pass
+
+        return results
+
+    def search_and_read(self, query, deep=True):
+        """Legacy sequential method for compatibility."""
+        results = self.search(query)
+        if deep:
+            for r in results[:2]:
+                content = self.scrape_page(r["url"])
+                if content:
+                    r["full_content"] = content
+        return results