Update multimodal_module.py

multimodal_module.py

@@ -1,1303 +1,806 @@
-# multimodal_superagent.py
+# multimodular_modul version 7.0.py
 """
-Multimodal SuperAgent v6.0 — Single-file master implementation
-COPY & PASTE into multimodal_superagent.py
-
-WHAT'S NEW vs v5.0 (upgrade, not a rewrite):
-- Close-to-Human Brain (CHB) layer: the only module allowed to speak to the user.
-  - Gathers evidence from memory, web, tools, models.
-  - Verifies claims with a Versioned Fact Store (VFS).
-  - Resolves contradictions (old vs new knowledge).
-  - Computes confidence; only answers when >= 0.90. Otherwise returns the certain subset.
-- Versioned Fact Store (JSON): timelines of facts, supersedes history, provenance & citations.
-- Confidence Scorer: reliability + recency + agreement + self-consistency + retrieval strength – contradictions.
-- Web search: local HTML scraping with 24h cache (title/url/snippet<=100 chars). No paid tokens needed.
-- Multimodal perception glue:
-  - Emojis/emotion tags for text (lightweight).
-  - Image caption hook (BLIP/ViT if configured) → stored as evidence.
-  - Video analyzer (keyframes). Basic FPS “interpolation” without external binaries (frame duplication).
-- Safer math & schema checks during verification (SymPy + simple validators).
-- Long-term memory, KG, agents, image & voice pipelines retained and improved with better error paths.
-- Citations and confidence optionally attached to replies (text mode).
-- Runs CPU or GPU (auto-detect). No hard dependencies on cloud APIs.
+Multimodular Module — Multimodal SuperAgent v7.0
+Upgrade from v6.0 -> v7.0 (in-place upgrade)
+
+Features added:
+- CHB (Close-to-Human Brain) is universal "middle-man" for all inputs/outputs.
+- User-driven retrieval plan flow (AI generates queries; client fetches; submits results).
+- Opportunistic Creative Skill Vault (media & text benchmarks).
+- Versioned Fact Store (VFS) enhancements: freshness, controversy scoring, provenance.
+- Real-time Global Brain (facts, skills, media) sync via WebSocket + optional HuggingFace dataset push.
+- Automatic local backups (JSON + SQLite), downloadable via API endpoint.
+- Self-upgrading modules: safe, signed modules can be auto-integrated into runtime.
+  * Auto-exec only when cryptographic signature verification + sandbox available (wasmtime or subprocess sandboxes).
+  * If signing or sandbox not present, modules are stored but not auto-executed.
+- Universal FastAPI endpoints and CLI demo preserved from v6 with additions.
+
+Security & Safety:
+- No CAPTCHA/TLS evasion. Respect robots.txt & user-driven retrieval model.
+- Self-upgrade requires signature verification (env GLOBAL_SYNC_SIGNING_PUBKEY).
+- Default: local-only sync. Enable cloud via env variables (HUGGINGFACE_TOKEN, GLOBAL_SYNC_REPO).
+- Media sync allowed; personal/private data must be filtered before upload.
 """
 
 from __future__ import annotations
-import os, sys, time, json, uuid, shutil, tempfile, subprocess, logging, math, hashlib, re
+import os, sys, json, time, uuid, shutil, tempfile, hashlib, base64, logging
 from dataclasses import dataclass, field, asdict
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple
+from pathlib import Path
+import threading
+import sqlite3
+import zipfile
+import hmac
+import hashlib
 import asyncio
 
-# --- lazy imports for heavy libs; imported when needed ---
+# --------------------------
+# Optional deps (feature unlocks)
+# --------------------------
+# pip install fastapi uvicorn pydantic requests websockets python-multipart cryptography wasmtime
 try:
-    import torch
+    import fastapi
+    from fastapi import FastAPI, UploadFile, File, Form
+    from fastapi.responses import FileResponse, JSONResponse
+    from pydantic import BaseModel
+    FASTAPI_AVAILABLE = True
 except Exception:
-    torch = None
+    FASTAPI_AVAILABLE = False
+
 try:
-    import numpy as np
+    import requests
 except Exception:
-    np = None
-
-# Logging
-logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
-logger = logging.getLogger("SuperAgent")
-
-# ---------------------------
-# CONFIG — edit BEFORE running
-# ---------------------------
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-TMP_DIR = os.getenv("SUPERAGENT_TMP", os.path.join(BASE_DIR, "tmp"))
-CACHE_DIR = os.getenv("SUPERAGENT_CACHE", os.path.join(BASE_DIR, "model_cache"))
-os.makedirs(TMP_DIR, exist_ok=True)
-os.makedirs(CACHE_DIR, exist_ok=True)
-
-# Central configuration: change model paths here or provide config.yaml and set env SUPERAGENT_CONFIG
-MODEL_CONFIG = {
-    "device": "auto",  # "auto" | "cpu" | "cuda"
-    "llm": {  # prefer local HF model id or path; optionally set 'backend' to 'llamacpp'
-        "model_path": None,         # e.g., "/path/to/Mistral-7B-Instruct"
-        "backend": "transformers",  # "transformers" | "llamacpp"
-        "quantize": None,           # e.g., "bitsandbytes" or GGUF for llamacpp
-    },
-    "embedder": "sentence-transformers/all-MiniLM-L6-v2",
-    "faiss_index_dir": os.path.join(CACHE_DIR, "faiss_index"),
-    "sdxl_base": None,        # e.g., "/path/to/sdxl-base"
-    "sdxl_refiner": None,     # optional
-    "sdxl_inpaint": None,     # optional
-    "blip_caption": None,     # e.g., "Salesforce/blip-image-captioning-base" (optional)
-    "piper_binary": "/usr/local/bin/piper",  # optional
-    "piper_voice": None,                     # optional
-    "allow_web_search": False,               # opt-in
-    "safety_blocklist": ["terror", "explosive", "harm"],
-    "knowledge_graph_path": os.path.join(CACHE_DIR, "kg.json"),
-    "memory_persist": True,
-    "memory_file": os.path.join(CACHE_DIR, "longterm_memory.json"),
-    "embed_dim": 384,  # MiniLM-L6-v2
-    "auto_learn": True,
-    "vfs_file": os.path.join(CACHE_DIR, "versioned_facts.json"),
-    "web_cache_file": os.path.join(CACHE_DIR, "webcache.json"),
-    "chb_min_confidence": 0.90,
-    "max_memory_items": 10000,
-    "kg_max_nodes": 500,
-    "short_memory_turns": 50,
-}
-
-# ---------------------------
-# Helpers
-# ---------------------------
-def uid(prefix="asset"):
-    return f"{prefix}_{uuid.uuid4().hex[:8]}"
-
-def tmp_path(suffix=""):
-    p = os.path.join(TMP_DIR, f"{uuid.uuid4().hex}{suffix}")
-    os.makedirs(os.path.dirname(p), exist_ok=True)
-    return p
-
-def cleanup(*paths):
-    for p in paths:
-        try:
-            if not p: continue
-            if os.path.isfile(p): os.remove(p)
-            elif os.path.isdir(p): shutil.rmtree(p)
-        except Exception as e:
-            logger.debug("cleanup failed %s: %s", p, e)
+    requests = None
 
-def now_ts():
-    return float(time.time())
-
-# ---------------------------
-# Provenance & Watermark
-# ---------------------------
-@dataclass
-class Provenance:
-    model: str
-    version: str
-    synthetic: bool = True
-    consent: Optional[str] = None
-    created_at: float = field(default_factory=now_ts)
-    extra: Dict[str, Any] = field(default_factory=dict)
-
-class ProvenanceManager:
-    def attach(self, filepath: str, meta: Union[Provenance, dict]):
-        try:
-            meta_out = asdict(meta) if isinstance(meta, Provenance) else meta
-            with open(filepath + ".prov.json", "w", encoding="utf-8") as f:
-                json.dump(meta_out, f, indent=2, ensure_ascii=False)
-        except Exception as e:
-            logger.warning("provenance attach failed: %s", e)
-
-    def watermark_audio(self, wav_bytes: bytes) -> bytes:
-        # Non-destructive placeholder; does not alter content.
-        return wav_bytes
-
-    def watermark_image(self, pil_img):
-        # Return as-is. Replace with robust watermark if desired.
-        return pil_img
-
-# ---------------------------
-# Safety & Consent
-# ---------------------------
-@dataclass
-class ConsentRecord:
-    user_id: str
-    attestation: str
-    ts: float = field(default_factory=now_ts)
-
-class SafetyManager:
-    def __init__(self, blocklist=None):
-        self.blocklist = blocklist or MODEL_CONFIG["safety_blocklist"]
-        self.consent_log: List[ConsentRecord] = []
-
-    def record_consent(self, user_id: str, attestation: str) -> str:
-        token = hashlib.sha256(f"{user_id}-{attestation}-{time.time()}".encode()).hexdigest()
-        self.consent_log.append(ConsentRecord(user_id=user_id, attestation=attestation))
-        return token
-
-    def is_allowed(self, text: str) -> bool:
-        t = (text or "").lower()
-        return not any(b in t for b in self.blocklist)
-
-    def check_public_figure(self, embedding) -> bool:
-        # Placeholder: implement real similarity against protected embeddings if available
-        return False
-
-# ---------------------------
-# Short-term memory (conversation buffer)
-# ---------------------------
-class ShortTermMemory:
-    def __init__(self, max_turns=None):
-        self.max_turns = max_turns or MODEL_CONFIG["short_memory_turns"]
-        self.store: Dict[str, List[Dict[str,Any]]] = {}
-
-    def push(self, user_id: str, role: str, content: Any):
-        k = str(user_id)
-        arr = self.store.setdefault(k, [])
-        arr.append({"ts":now_ts(), "role":role, "content":content})
-        if len(arr) > self.max_turns:
-            self.store[k] = arr[-self.max_turns:]
+try:
+    import websockets
+except Exception:
+    websockets = None
 
-    def recent(self, user_id: str, k=10):
-        return self.store.get(str(user_id), [])[-k:]
+try:
+    from cryptography.hazmat.primitives import serialization, hashes
+    from cryptography.hazmat.primitives.asymmetric import padding
+    CRYPTO_AVAILABLE = True
+except Exception:
+    CRYPTO_AVAILABLE = False
 
-# ---------------------------
-# Long-term memory (FAISS + Sentence-Transformers)
-# ---------------------------
-class LongTermMemory:
-    def __init__(self, index_dir=None, embed_model_name=None):
-        self.index_dir = index_dir or MODEL_CONFIG["faiss_index_dir"]
-        os.makedirs(self.index_dir, exist_ok=True)
-        self.memfile = MODEL_CONFIG.get("memory_file")
-        self.embed_model_name = embed_model_name or MODEL_CONFIG["embedder"]
-        self.data: List[Dict[str,Any]] = []
-        self.index = None
-        self.embedder = None
-        self.dim = MODEL_CONFIG.get("embed_dim", 384)
-        self._load()
+try:
+    import wasmtime
+    WASM_AVAILABLE = True
+except Exception:
+    WASM_AVAILABLE = False
 
-    def _load(self):
-        if os.path.exists(self.memfile):
-            try:
-                with open(self.memfile, "r", encoding="utf-8") as f:
-                    self.data = json.load(f)
-            except Exception:
-                self.data = []
-        # load faiss if available
-        try:
-            import faiss  # type: ignore
-            idx_path = os.path.join(self.index_dir, "index.faiss")
-            if os.path.exists(idx_path):
-                try:
-                    self.index = faiss.read_index(idx_path)
-                except Exception:
-                    self.index = None
+# --------------------------
+# Logging
+# --------------------------
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+log = logging.getLogger("Multimodular_v7")
+
+# --------------------------
+# Config & Paths (edit env or constants)
+# --------------------------
+BASE_DIR = Path(os.getenv("MM_BASE_DIR", Path(__file__).parent.resolve()))
+DATA_DIR = Path(os.getenv("MM_DATA_DIR", BASE_DIR / "mm_data"))
+BACKUP_DIR = Path(os.getenv("MM_BACKUP_DIR", DATA_DIR / "backups"))
+TMP_DIR = Path(os.getenv("MM_TMP_DIR", BASE_DIR / "tmp"))
+CACHE_DIR = Path(os.getenv("MM_CACHE_DIR", BASE_DIR / "cache"))
+for d in (DATA_DIR, BACKUP_DIR, TMP_DIR, CACHE_DIR):
+    d.mkdir(parents=True, exist_ok=True)
+
+# Global sync config
+GLOBAL_SYNC_ENABLED = os.getenv("MM_GLOBAL_SYNC_ENABLED", "false").lower() in ("1","true","yes")
+HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_TOKEN", None)
+GLOBAL_SYNC_REPO = os.getenv("GLOBAL_SYNC_REPO", None)  # e.g., "username/mm_global_brain"
+GLOBAL_SYNC_SIGNING_PUBKEY = os.getenv("GLOBAL_SYNC_SIGNING_PUBKEY", None)  # PEM public key for verifying modules
+REALTIME_WS_PORT = int(os.getenv("MM_WS_PORT", "8765"))
+
+# Auto-upgrade strictness: require signature & sandbox for auto-exec
+AUTO_UPGRADE_REQUIRE_SIGN = True
+AUTO_UPGRADE_REQUIRE_SANDBOX = True
+
+# Backups
+BACKUP_RETENTION = int(os.getenv("MM_BACKUP_RETENTION", "30"))  # keep last N backups
+
+# CHB confidence threshold
+CHB_MIN_CONFIDENCE = float(os.getenv("CHB_MIN_CONFIDENCE", "0.85"))
+
+# --------------------------
+# Utilities
+# --------------------------
+def uid(prefix="id"):
+    return f"{prefix}_{uuid.uuid4().hex[:10]}"
+
+def now_iso():
+    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+
+def sha256_b64(data: bytes) -> str:
+    return base64.urlsafe_b64encode(hashlib.sha256(data).digest()).decode()
+
+def write_json(path: Path, data: Any):
+    tmp = path.with_suffix(".tmp")
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+    tmp.replace(path)
+
+def read_json(path: Path, default=None):
+    if not path.exists():
+        return default
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return default
+
+# --------------------------
+# Database: local SQLite wrapper + JSON mirror
+# --------------------------
+class LocalDB:
+    def __init__(self, path: Path):
+        self.path = path
+        self.conn = sqlite3.connect(str(self.path))
+        self._init()
+        self.lock = threading.Lock()
+    def _init(self):
+        cur = self.conn.cursor()
+        cur.execute("""CREATE TABLE IF NOT EXISTS facts (
+                        id TEXT PRIMARY KEY,
+                        claim TEXT,
+                        value TEXT,
+                        confidence REAL,
+                        sources TEXT,
+                        first_seen TEXT,
+                        last_seen TEXT,
+                        controversy REAL,
+                        staleness REAL
+                    )""")
+        cur.execute("""CREATE TABLE IF NOT EXISTS skills (
+                        id TEXT PRIMARY KEY,
+                        tag TEXT,
+                        kind TEXT,
+                        meta TEXT,
+                        score REAL,
+                        added_at TEXT,
+                        synced INTEGER DEFAULT 0
+                    )""")
+        cur.execute("""CREATE TABLE IF NOT EXISTS modules (
+                        id TEXT PRIMARY KEY,
+                        name TEXT,
+                        code TEXT,
+                        meta TEXT,
+                        verified INTEGER DEFAULT 0,
+                        autointegrated INTEGER DEFAULT 0,
+                        added_at TEXT
+                    )""")
+        cur.execute("""CREATE TABLE IF NOT EXISTS backups (
+                        id TEXT PRIMARY KEY,
+                        path TEXT,
+                        created_at TEXT
+                    )""")
+        self.conn.commit()
+    def upsert_fact(self, claim, value, confidence, sources, controversy=0.0, staleness=0.0):
+        fid = sha256_b64(claim.encode())[:32]
+        now = now_iso()
+        with self.lock:
+            cur = self.conn.cursor()
+            cur.execute("SELECT id FROM facts WHERE id=?", (fid,))
+            if cur.fetchone():
+                cur.execute("""UPDATE facts SET value=?, confidence=?, sources=?, last_seen=?, controversy=?, staleness=? WHERE id=?""",
+                            (value, float(confidence), json.dumps(sources), now, float(controversy), float(staleness), fid))
             else:
-                self.index = None
-        except Exception:
-            self.index = None
-
-    def _save(self):
-        with open(self.memfile, "w", encoding="utf-8") as f:
-            json.dump(self.data, f, indent=2, ensure_ascii=False)
-        if self.index is not None:
+                cur.execute("""INSERT INTO facts (id,claim,value,confidence,sources,first_seen,last_seen,controversy,staleness)
+                               VALUES (?,?,?,?,?,?,?,?,?)""",
+                            (fid, claim, value, float(confidence), json.dumps(sources), now, now, float(controversy), float(staleness)))
+            self.conn.commit()
+        return fid
+    def add_skill(self, tag, kind, meta, score):
+        sid = uid("skill")
+        now = now_iso()
+        with self.lock:
+            self.conn.execute("INSERT INTO skills (id,tag,kind,meta,score,added_at) VALUES (?,?,?,?,?,?)",
+                              (sid, tag, kind, json.dumps(meta), float(score), now))
+            self.conn.commit()
+        return sid
+    def add_module(self, name, code, meta, verified=0, autointegrated=0):
+        mid = uid("mod")
+        now = now_iso()
+        with self.lock:
+            self.conn.execute("INSERT INTO modules (id,name,code,meta,verified,autointegrated,added_at) VALUES (?,?,?,?,?,?,?)",
+                              (mid, name, code, json.dumps(meta), int(verified), int(autointegrated), now))
+            self.conn.commit()
+        return mid
+    def list_facts(self):
+        cur = self.conn.cursor(); cur.execute("SELECT * FROM facts"); rows=cur.fetchall()
+        cols=[c[0] for c in cur.description]
+        return [dict(zip(cols, r)) for r in rows]
+    def list_skills(self):
+        cur = self.conn.cursor(); cur.execute("SELECT * FROM skills"); rows=cur.fetchall()
+        cols=[c[0] for c in cur.description]
+        return [dict(zip(cols, r)) for r in rows]
+    def list_modules(self):
+        cur = self.conn.cursor(); cur.execute("SELECT * FROM modules"); rows=cur.fetchall()
+        cols=[c[0] for c in cur.description]
+        return [dict(zip(cols, r)) for r in rows]
+    def mark_module_verified(self, module_id, verified=1):
+        with self.lock:
+            self.conn.execute("UPDATE modules SET verified=? WHERE id=?", (int(verified), module_id))
+            self.conn.commit()
+    def mark_module_autointegrated(self, module_id, val=1):
+        with self.lock:
+            self.conn.execute("UPDATE modules SET autointegrated=? WHERE id=?", (int(val), module_id))
+            self.conn.commit()
+    def add_backup(self, path):
+        bid = uid("bak")
+        now = now_iso()
+        with self.lock:
+            self.conn.execute("INSERT INTO backups (id,path,created_at) VALUES (?,?,?)", (bid, str(path), now))
+            self.conn.commit()
+        self._prune_backups()
+        return bid
+    def _prune_backups(self):
+        cur = self.conn.cursor(); cur.execute("SELECT id,path,created_at FROM backups ORDER BY created_at DESC")
+        rows = cur.fetchall()
+        if len(rows) <= BACKUP_RETENTION: return
+        for r in rows[BACKUP_RETENTION:]:
+            pid, p, _ = r
             try:
-                import faiss  # type: ignore
-                faiss.write_index(self.index, os.path.join(self.index_dir, "index.faiss"))
+                if os.path.exists(p): os.remove(p)
             except Exception:
                 pass
-
-    def _get_embedder(self):
-        if self.embedder is None:
-            try:
-                from sentence_transformers import SentenceTransformer
-                self.embedder = SentenceTransformer(self.embed_model_name)
-            except Exception as e:
-                logger.debug("embedder load failed: %s", e)
-                self.embedder = None
-        return self.embedder
-
-    def _ensure_index(self, dim:int):
-        if self.index is None:
-            try:
-                import faiss  # type: ignore
-                self.index = faiss.IndexFlatL2(dim)
-            except Exception:
-                self.index = None
-
-    def add(self, user_id: str, text: str, kind: str="turn"):
-        if not text: return
-        # prune if oversized
-        if len(self.data) >= MODEL_CONFIG["max_memory_items"]:
-            self.data = self.data[int(0.1*len(self.data)):]  # drop oldest 10%
-        item = {"id": uid("m"), "user_id": str(user_id), "text": text, "kind": kind, "ts": now_ts()}
-        self.data.append(item)
-        emb = self._get_embedder()
-        if emb is not None:
-            vec = emb.encode([text])
-            import numpy as _np
-            self._ensure_index(vec.shape[1])
-            if self.index is not None:
-                self.index.add(_np.asarray(vec, dtype="float32"))
-        if MODEL_CONFIG.get("memory_persist"):
-            self._save()
-
-    def search(self, query: str, top_k=5) -> List[Dict[str,Any]]:
-        if not self.data:
-            return []
-        emb = self._get_embedder()
-        if emb is None or self.index is None:
-            # fallback naive search
-            q = (query or "").lower()
-            scored=[]
-            for i,m in enumerate(self.data):
-                s = sum(1 for tok in q.split() if tok in m["text"].lower())
-                if s>0: scored.append((s,i))
-            scored.sort(reverse=True)
-            return [self.data[i] for _,i in scored[:top_k]]
-        qv = emb.encode([query])
-        import numpy as _np
-        D,I = self.index.search(_np.asarray(qv, dtype="float32"), min(top_k, len(self.data)))
-        res=[]
-        used=set()
-        for idx in I[0]:
-            if 0 <= idx < len(self.data) and int(idx) not in used:
-                used.add(int(idx))
-                res.append(self.data[int(idx)])
-        return res
-
-    def export_all(self):
-        return {"count": len(self.data), "items": self.data}
-
-    def import_bulk(self, items: List[Dict[str,Any]]):
-        self.data = items or []
-        emb = self._get_embedder()
-        if emb is not None and self.data:
-            vecs = emb.encode([m["text"] for m in self.data])
-            import numpy as _np
-            self._ensure_index(vecs.shape[1])
-            if self.index is not None:
-                self.index.reset()
-                self.index.add(_np.asarray(vecs, dtype="float32"))
-        self._save()
-
-# ---------------------------
-# Knowledge Graph (simple JSON triples)
-# ---------------------------
-class KnowledgeGraph:
-    def __init__(self, path=None):
-        self.path = path or MODEL_CONFIG["knowledge_graph_path"]
-        self.graph = {}
-        self._load()
-
-    def _load(self):
-        if os.path.exists(self.path):
-            try:
-                with open(self.path,"r",encoding="utf-8") as f:
-                    self.graph = json.load(f)
-            except Exception:
-                self.graph = {}
-
-    def add_fact(self, subj: str, pred: str, obj: str):
-        # cap nodes to avoid unbounded growth
-        if len(self.graph) > MODEL_CONFIG["kg_max_nodes"]:
-            # basic prune: drop oldest 10% of keys by insertion order
-            keys = list(self.graph.keys())
-            for k in keys[:max(1, len(keys)//10)]:
-                self.graph.pop(k, None)
-        k = f"{subj}::{pred}"
-        self.graph.setdefault(k, []).append(obj)
-        self._save()
-
-    def query(self, subj: str, pred: str):
-        return self.graph.get(f"{subj}::{pred}", [])
-
-    def _save(self):
+            self.conn.execute("DELETE FROM backups WHERE id=?", (pid,))
+        self.conn.commit()
+
+# --------------------------
+# VFS and Creative Skill Vault (JSON + SQLite)
+# --------------------------
+class VFS:
+    def __init__(self, db: LocalDB):
+        self.db = db
+    def store_fact(self, claim:str, value:str, sources:List[Dict[str,Any]], confidence:float, controversy:float=0.0):
+        # staleness computed from source dates (simple)
+        staleness = 0.0
+        for s in (sources or []):
+            dt = s.get("date")
+            if dt:
+                try:
+                    # naive parse as ISO; compute days
+                    t = time.mktime(time.strptime(dt[:19], "%Y-%m-%dT%H:%M:%S"))
+                    age_days = max(0, (time.time() - t)/86400.0)
+                    staleness = max(staleness, min(1.0, age_days/365.0))
+                except Exception:
+                    continue
+        fid = self.db.upsert_fact(claim, value, confidence, sources, controversy, staleness)
+        return fid
+    def query(self, q:str):
+        # naive substring search
+        res = self.db.list_facts()
+        qlow = q.lower()
+        return [r for r in res if qlow in (r.get("claim") or "").lower() or qlow in (r.get("value") or "").lower()]
+
+class CreativeSkillVault:
+    def __init__(self, db: LocalDB):
+        self.db = db
+    def add_benchmark(self, tag:str, kind:str, meta:Dict[str,Any], score:float):
+        sid = self.db.add_skill(tag, kind, meta, score)
+        return sid
+    def top_by_tag(self, tag:str, k:int=5):
+        all_skills = self.db.list_skills()
+        filtered = [s for s in all_skills if s.get("tag")==tag]
+        filtered.sort(key=lambda x: x.get("score",0), reverse=True)
+        return filtered[:k]
+
+# --------------------------
+# Global Sync: Hugging Face push & WebSocket real-time (simple)
+# --------------------------
+class GlobalSync:
+    def __init__(self, db: LocalDB, hf_token:Optional[str]=None, repo:Optional[str]=None):
+        self.db = db
+        self.hf_token = hf_token
+        self.repo = repo
+        self.ws_clients = set()
+        self.ws_server_task = None
+        self.loop = None
+        self.lock = threading.Lock()
+    # --- push facts/skills package to Hugging Face dataset via simple HTTP (requires token & repo)
+    def push_to_hf(self, package:Dict[str,Any]) -> Tuple[bool,str]:
+        if not (self.hf_token and self.repo and requests):
+            return False, "huggingface not configured or requests missing"
+        # Minimal implementation: upload JSON file to HF repo via API
         try:
-            with open(self.path,"w",encoding="utf-8") as f:
-                json.dump(self.graph, f, indent=2, ensure_ascii=False)
+            url = f"https://huggingface.co/api/repos/create"
+            # Note: full implementation requires use of hf_hub or dataset APIs; here we do a simple placeholder
+            # We recommend using huggingface_hub library in production.
+            return False, "HF push requires huggingface_hub implementation; configure HF client"
         except Exception as e:
-            logger.debug("kg save failed: %s", e)
-
-# ---------------------------
-# Versioned Fact Store (VFS)
-# ---------------------------
-class VersionedFactStore:
-    """
-    Stores facts with versioning & provenance.
-    Fact schema:
-      {
-        "id": str, "claim": str, "value": str, "scope": str|None,
-        "first_seen": ts, "verified_at": ts, "confidence": float,
-        "sources": [{"type": "web|memory|kg|vision|audio|manual", "ref": str, "title": str|None, "time": ts}],
-        "supersedes": str|None, "valid_from": ts|None, "valid_to": ts|None
-      }
-    """
-    def __init__(self, path=None):
-        self.path = path or MODEL_CONFIG["vfs_file"]
-        self.facts: List[Dict[str,Any]] = []
-        self._load()
-
-    def _load(self):
-        if os.path.exists(self.path):
+            return False, str(e)
+    # --- broadcast to connected WebSocket clients (realtime)
+    async def ws_broadcast(self, message:Dict[str,Any]):
+        if websockets is None:
+            return
+        data = json.dumps(message)
+        clients = list(self.ws_clients)
+        for ws in clients:
             try:
-                with open(self.path, "r", encoding="utf-8") as f:
-                    self.facts = json.load(f)
+                await ws.send(data)
             except Exception:
-                self.facts = []
-
-    def _save(self):
-        try:
-            with open(self.path, "w", encoding="utf-8") as f:
-                json.dump(self.facts, f, indent=2, ensure_ascii=False)
-        except Exception as e:
-            logger.debug("vfs save failed: %s", e)
-
-    def add_or_update(self, claim: str, value: str, sources: List[Dict[str,Any]], confidence: float,
-                      scope: Optional[str]=None, supersedes: Optional[str]=None,
-                      valid_from: Optional[float]=None, valid_to: Optional[float]=None) -> Dict[str,Any]:
-        new_id = uid("fact")
-        rec = {
-            "id": new_id, "claim": claim, "value": value, "scope": scope,
-            "first_seen": now_ts(), "verified_at": now_ts(), "confidence": float(confidence),
-            "sources": sources or [], "supersedes": supersedes,
-            "valid_from": valid_from, "valid_to": valid_to
-        }
-        self.facts.append(rec)
-        self._save()
-        return rec
-
-    def find(self, claim: str) -> List[Dict[str,Any]]:
-        c = claim.strip().lower()
-        return [f for f in self.facts if f.get("claim","").strip().lower()==c]
-
-    def latest(self, claim: str) -> Optional[Dict[str,Any]]:
-        items = self.find(claim)
-        if not items: return None
-        # return the most recent verified_at
-        return sorted(items, key=lambda x: x.get("verified_at", 0), reverse=True)[0]
-
-    def all(self) -> List[Dict[str,Any]]:
-        return self.facts
-
-# ---------------------------
-# Web search (HTML scrape + 24h cache)
-# ---------------------------
-class WebSearch:
-    def __init__(self, enabled: bool=False, cache_file: Optional[str]=None):
-        self.enabled = bool(enabled)
-        self.cache_file = cache_file or MODEL_CONFIG["web_cache_file"]
-        self.cache = {}
-        self._load_cache()
-
-    def _load_cache(self):
-        if os.path.exists(self.cache_file):
+                try:
+                    self.ws_clients.remove(ws)
+                except Exception:
+                    pass
+    # --- start a simple websocket server to accept other CHBs / clients that want live updates
+    def start_ws_server(self, host="0.0.0.0", port=REALTIME_WS_PORT):
+        if websockets is None:
+            log.warning("websockets library missing; realtime sync disabled")
+            return
+        async def handler(websocket, path):
+            log.info("WS client connected")
+            self.ws_clients.add(websocket)
             try:
-                with open(self.cache_file, "r", encoding="utf-8") as f:
-                    self.cache = json.load(f)
+                async for msg in websocket:
+                    # accept 'ping' or 'submit' messages
+                    try:
+                        data = json.loads(msg)
+                        typ = data.get("type")
+                        if typ == "submit_skill":
+                            payload = data.get("payload")
+                            # minimal processing: store skill locally and broadcast
+                            tag = payload.get("tag","global")
+                            kind = payload.get("kind","image")
+                            meta = payload.get("meta",{})
+                            score = float(payload.get("score", 0.5))
+                            self.db.add_skill(tag, kind, meta, score)
+                            await self.ws_broadcast({"type":"skill_added","tag":tag,"kind":kind,"meta":meta,"score":score})
+                    except Exception:
+                        pass
             except Exception:
-                self.cache = {}
-
-    def _save_cache(self):
+                pass
+            finally:
+                try: self.ws_clients.remove(websocket)
+                except Exception: pass
+                log.info("WS client disconnected")
+        log.info("Starting WebSocket server on %s:%d", host, port)
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+        start_server = websockets.serve(handler, host, port)
+        self.ws_server_task = self.loop.run_until_complete(start_server)
         try:
-            with open(self.cache_file, "w", encoding="utf-8") as f:
-                json.dump(self.cache, f, indent=2, ensure_ascii=False)
+            self.loop.run_forever()
         except Exception:
             pass
 
-    def search(self, query: str, max_results: int=3) -> List[Dict[str,str]]:
-        if not self.enabled:
-            return []
-        key = hashlib.sha1(query.encode()).hexdigest()
-        # 24h TTL
-        if key in self.cache and (now_ts() - self.cache[key]["ts"]) < 86400:
-            return self.cache[key]["hits"][:max_results]
+    def run_ws_in_thread(self, host="0.0.0.0", port=REALTIME_WS_PORT):
+        t = threading.Thread(target=self.start_ws_server, args=(host,port), daemon=True)
+        t.start()
+        return t
+
+# --------------------------
+# ModuleManager: Verify & Sandbox auto-integration of learned modules
+# --------------------------
+class ModuleManager:
+    def __init__(self, db: LocalDB, signing_pubkey_pem:Optional[str]=None):
+        self.db = db
+        self.signing_pubkey_pem = signing_pubkey_pem
+        self.sandbox_available = WASM_AVAILABLE  # prefer WASM sandbox if available
+        self.lock = threading.Lock()
+
+    def verify_signature(self, code: bytes, signature_b64: str) -> bool:
+        if not (CRYPTO_AVAILABLE and self.signing_pubkey_pem):
+            log.warning("Crypto or public key not available, cannot verify signature")
+            return False
         try:
-            import requests
-            from bs4 import BeautifulSoup
-            url = f"https://duckduckgo.com/html/?q={requests.utils.quote(query)}"
-            headers = {"User-Agent":"Mozilla/5.0"}
-            r = requests.get(url, headers=headers, timeout=8)
-            soup = BeautifulSoup(r.text, "html.parser")
-            results=[]
-            # capture title/link/snippet (≤100 chars)
-            containers = soup.select(".result")[:max_results]
-            for c in containers:
-                a = c.select_one(".result__a")
-                s = c.select_one(".result__snippet")
-                title = a.get_text(strip=True) if a else ""
-                link = a.get("href") if a else ""
-                snippet = (s.get_text(" ", strip=True) if s else "")[:100]
-                if title and link:
-                    results.append({"title": title, "link": link, "snippet": snippet})
-            self.cache[key] = {"ts": now_ts(), "hits": results}
-            self._save_cache()
-            return results
+            pub = serialization.load_pem_public_key(self.signing_pubkey_pem.encode())
+            sig = base64.b64decode(signature_b64)
+            pub.verify(sig, code, padding.PKCS1v15(), hashes.SHA256())
+            return True
         except Exception as e:
-            logger.debug("websearch failed: %s", e)
-            return []
+            log.warning("signature verification failed: %s", e)
+            return False
 
-# ---------------------------
-# LLM wrapper (transformers / llama.cpp)
-# ---------------------------
-class LocalLLM:
-    def __init__(self, model_path=None, backend="transformers", device="auto", quantize=None):
-        self.model_path = model_path
-        self.backend = backend
-        self.quantize = quantize
-        self.device = device if device!="auto" else ("cuda" if torch and torch.cuda.is_available() else "cpu")
-        self.model = None
-        self.tokenizer = None
-        self._loaded = False
+    def sandbox_run_wasm(self, wasm_bytes: bytes, func_name: str="run", inputs: Optional[dict]=None, timeout: int=5) -> Tuple[bool,str]:
+        if not WASM_AVAILABLE:
+            return False, "wasm runtime not available"
+        try:
+            # create store & module
+            engine = wasmtime.Engine()
+            module = wasmtime.Module(engine, wasm_bytes)
+            store = wasmtime.Store(engine)
+            instance = wasmtime.Instance(store, module, [])
+            # This is a very conservative pattern — real WASM modules need standard interface; here we just attempt safe run if exposes memory/free functions
+            # For safety, we do not invoke arbitrary functions unless module authors follow the expected interface
+            # We'll attempt to call an exported function named 'run' that returns int
+            if hasattr(instance.exports, func_name):
+                fn = instance.exports.__getattr__(func_name)
+                try:
+                    res = fn()
+                    return True, f"wasm-run-res:{res}"
+                except Exception as e:
+                    return False, f"wasm-run-exc:{e}"
+            else:
+                return False, "wasm module lacks 'run' export"
+        except Exception as e:
+            return False, f"wasm-failed:{e}"
+
+    def sandbox_run_subprocess(self, code_str: str, timeout: int=5) -> Tuple[bool,str]:
+        # Very limited subprocess sandbox: write file, run in subprocess with restricted env and timeout.
+        # NOTE: this is not fully secure against malicious code. Use real OS-level sandboxing for production.
+        tmp = Path(TMP_DIR) / f"module_{uid()}.py"
+        tmp.write_text(code_str, encoding="utf-8")
+        import subprocess, shlex
+        try:
+            p = subprocess.run([sys.executable, str(tmp)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, check=False)
+            out = p.stdout.decode()[:4000]
+            err = p.stderr.decode()[:2000]
+            return True, out + ("\nERR:\n" + err if err else "")
+        except subprocess.TimeoutExpired:
+            return False, "timeout"
+        except Exception as e:
+            return False, f"exec-error:{e}"
+        finally:
+            try: tmp.unlink()
+            except Exception: pass
 
-    def load(self):
-        if self._loaded: return
-        if self.backend == "transformers":
+    def integrate_module(self, name: str, code: str, signature_b64: Optional[str]=None, autointegrate: bool=True) -> Dict[str,Any]:
+        # Store module first
+        meta = {"name": name, "signature": bool(signature_b64), "autointegrate": bool(autointegrate)}
+        mid = self.db.add_module(name, code, meta, verified=0, autointegrated=0)
+        # Verify signature if present and required
+        if AUTO_UPGRADE_REQUIRE_SIGN:
+            if not signature_b64 or not self.verify_signature(code.encode(), signature_b64):
+                return {"ok": False, "reason": "signature_missing_or_invalid", "module_id": mid}
+        # Sandbox-run tests
+        ran_ok = False; run_info = None
+        if self.sandbox_available and AUTO_UPGRADE_REQUIRE_SANDBOX:
+            # expect code to be WASM base64 (prefer) or python code string. Detect if code is base64 wasm by heuristic.
             try:
-                from transformers import AutoModelForCausalLM, AutoTokenizer, AutoConfig
-                if not self.model_path:
-                    raise ValueError("MODEL_CONFIG['llm']['model_path'] not set")
-                logger.info("Loading transformers model %s", self.model_path)
-                self.tokenizer = AutoTokenizer.from_pretrained(self.model_path, cache_dir=CACHE_DIR)
-                # device_map="auto" keeps it simple; will use GPU if available
-                self.model = AutoModelForCausalLM.from_pretrained(
-                    self.model_path, device_map="auto", cache_dir=CACHE_DIR
-                )
-                self._loaded = True
+                # try decode base64, check for wasm magic
+                raw = base64.b64decode(code)
+                if raw[:4] == b"\x00asm":
+                    ok, info = self.sandbox_run_wasm(raw)
+                    ran_ok, run_info = ok, info
+                else:
+                    # treat as python source
+                    ok, info = self.sandbox_run_subprocess(code)
+                    ran_ok, run_info = ok, info
             except Exception as e:
-                logger.error("transformers LLM load failed: %s", e)
-                self._loaded = False
-        elif self.backend == "llamacpp":
-            try:
-                import llama_cpp
-                if not self.model_path:
-                    raise ValueError("MODEL_CONFIG['llm']['model_path'] not set")
-                self.model = llama_cpp.Llama(model_path=self.model_path)
-                self._loaded = True
-            except Exception as e:
-                logger.error("llamacpp load failed: %s", e)
-                self._loaded = False
+                ran_ok, run_info = False, f"sandbox-error:{e}"
         else:
-            logger.error("Unknown backend %s", self.backend)
-            self._loaded = False
-
-    def ready(self) -> bool:
-        return self._loaded
-
-    def generate(self, prompt: str, max_tokens: int=256, temperature: float=0.7) -> str:
-        if not self._loaded:
-            raise RuntimeError("LLM not loaded")
-        if self.backend == "transformers":
-            try:
-                inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
-                out = self.model.generate(
-                    **inputs, max_new_tokens=max_tokens, do_sample=True, temperature=temperature,
-                    pad_token_id=self.tokenizer.eos_token_id
-                )
-                return self.tokenizer.decode(out[0], skip_special_tokens=True)
-            except Exception as e:
-                logger.error("LLM generation failed: %s", e)
-                return "[llm-error]"
-        elif self.backend == "llamacpp":
-            try:
-                result = self.model.create(prompt=prompt, max_tokens=max_tokens, temperature=temperature)
-                if isinstance(result, dict) and "choices" in result:
-                    return result["choices"][0]["text"]
-                return str(result)
-            except Exception as e:
-                logger.error("llamacpp generation failed: %s", e)
-                return "[llm-error]"
+            # sandbox not available; do not autointegrate (security)
+            ran_ok, run_info = False, "sandbox-not-available"
+        # If sandboxed OK and autointegrate allowed, mark module autointegrated and (optionally) import into runtime
+        if ran_ok and autointegrate:
+            self.db.mark_module_verified(mid, verified=1)
+            # For safety, we will NOT exec arbitrary Python into this process.
+            # Instead, save module to disk as a safe package and mark autointegrated. A separate process can load it.
+            self.db.mark_module_autointegrated(mid, val=1)
+            return {"ok": True, "module_id": mid, "sandbox_result": run_info}
         else:
-            return "[backend-not-implemented]"
-
-# ---------------------------
-# Image engine (SDXL/diffusers + caption)
-# ---------------------------
-class ImageEngine:
-    def __init__(self, base=None, refiner=None, inpaint=None, blip=None, device="auto"):
-        self.paths = {"base": base, "refiner": refiner, "inpaint": inpaint, "blip": blip}
-        self.device = device if device!="auto" else ("cuda" if torch and torch.cuda.is_available() else "cpu")
-        self.pipe = None
-        self.refiner = None
-        self.inpaint_pipe = None
-        self.captioner = None
-
-    def load_pipes(self):
-        if self.pipe is None and self.paths.get("base"):
-            try:
-                from diffusers import StableDiffusionXLPipeline
-                dtype = torch.float16 if self.device=="cuda" else torch.float32
-                self.pipe = StableDiffusionXLPipeline.from_pretrained(self.paths["base"], torch_dtype=dtype, cache_dir=CACHE_DIR)
-                self.pipe.to(self.device)
-            except Exception as e:
-                logger.debug("sdxl base load failed: %s", e)
-                self.pipe = None
-        if self.refiner is None and self.paths.get("refiner"):
-            try:
-                from diffusers import StableDiffusionXLImg2ImgPipeline
-                dtype = torch.float16 if self.device=="cuda" else torch.float32
-                self.refiner = StableDiffusionXLImg2ImgPipeline.from_pretrained(self.paths["refiner"], torch_dtype=dtype, cache_dir=CACHE_DIR)
-                self.refiner.to(self.device)
-            except Exception as e:
-                logger.debug("refiner load failed: %s", e)
-                self.refiner = None
-        if self.inpaint_pipe is None and self.paths.get("inpaint"):
-            try:
-                from diffusers import StableDiffusionXLInpaintPipeline
-                dtype = torch.float16 if self.device=="cuda" else torch.float32
-                self.inpaint_pipe = StableDiffusionXLInpaintPipeline.from_pretrained(self.paths["inpaint"], torch_dtype=dtype, cache_dir=CACHE_DIR)
-                self.inpaint_pipe.to(self.device)
-            except Exception as e:
-                logger.debug("inpaint load failed: %s", e)
-                self.inpaint_pipe = None
-        if self.captioner is None and self.paths.get("blip"):
-            try:
-                from transformers import pipeline
-                device_idx = 0 if self.device=="cuda" else -1
-                self.captioner = pipeline("image-to-text", model=self.paths["blip"], device=device_idx, cache_dir=CACHE_DIR)
-            except Exception as e:
-                logger.debug("captioner load failed: %s", e)
-                self.captioner = None
-
-    def generate(self, prompt: str, negative: Optional[str]=None, steps:int=30, width:int=1024, height:int=1024, seed:Optional[int]=None):
-        self.load_pipes()
-        if not self.pipe:
-            raise RuntimeError("SDXL text2img pipeline unavailable. Provide path in MODEL_CONFIG['sdxl_base']")
-        generator = None
-        if torch and seed is not None:
-            generator = torch.Generator(device=self.device); generator.manual_seed(int(seed))
-        out = self.pipe(prompt=prompt, negative_prompt=negative or "", num_inference_steps=steps,
-                        width=width, height=height, generator=generator).images[0]
-        if self.refiner:
-            try:
-                out = self.refiner(prompt=prompt, image=out, strength=0.2).images[0]
-            except Exception:
-                pass
-        path = tmp_path(".png")
-        out.save(path)
-        return path
-
-    def inpaint(self, image_path: str, mask_path: Optional[str], prompt: str=""):
-        self.load_pipes()
-        if not self.inpaint_pipe:
-            raise RuntimeError("Inpaint pipeline not available.")
-        from PIL import Image
-        init = Image.open(image_path).convert("RGB")
-        mask = Image.open(mask_path).convert("L") if mask_path else Image.new("L", init.size, 255)
-        res = self.inpaint_pipe(prompt=prompt or " ", image=init, mask_image=mask, guidance_scale=7.5, num_inference_steps=30)
-        out = tmp_path("_inpaint.png"); res.images[0].save(out); return out
-
-    def caption(self, image_path: str) -> str:
-        self.load_pipes()
-        if not self.captioner: return ""
-        from PIL import Image
-        img = Image.open(image_path).convert("RGB")
-        try:
-            return self.captioner(img)[0].get("generated_text","")
-        except Exception:
-            return ""
-
-# ---------------------------
-# Video engine (keyframes) + functional FPS "interpolation"
-# ---------------------------
-class VideoEngine:
-    def __init__(self):
-        pass
-
-    def analyze(self, video_path: str, max_frames:int=6):
-        try:
-            import imageio
-            from PIL import Image
-            reader = imageio.get_reader(video_path)
-            total = reader.count_frames()
-            step = max(1, total // max_frames) if total else 1
-            frames=[]
-            for i in range(0,total,step):
+            return {"ok": False, "module_id": mid, "sandbox_result": run_info}
+
+# --------------------------
+# CHB — Universal middleman (upgrades v6 behaviour)
+# --------------------------
+class CHB:
+    def __init__(self, db: LocalDB, vfs: VFS, csv: CreativeSkillVault, module_mgr: ModuleManager, global_sync: GlobalSync):
+        self.db = db
+        self.vfs = vfs
+        self.csv = csv
+        self.module_mgr = module_mgr
+        self.global_sync = global_sync
+        # lightweight internal state
+        self.min_conf = CHB_MIN_CONFIDENCE
+
+    def perceive(self, incoming: Dict[str,Any]) -> Dict[str,Any]:
+        # normalize inputs (text,image,audio,video,plan_results)
+        kind = "text"
+        if incoming.get("image") or incoming.get("image_path"): kind="image"
+        if incoming.get("audio") or incoming.get("audio_path"): kind="audio"
+        if incoming.get("video") or incoming.get("video_path"): kind="video"
+        if incoming.get("plan_results"): kind="plan_results"
+        return {"kind": kind, "payload": incoming}
+
+    def plan(self, text: str) -> Dict[str,Any]:
+        # produce a user-driven retrieval plan
+        queries = []
+        queries.append({"q": text, "type":"web", "max_results":5})
+        # image/video heuristics
+        if any(k in text.lower() for k in ["image","design","render","photo","logo","illustration","concept"]):
+            queries.append({"q": text + " high quality concept art", "type":"image", "max_results":8})
+        if any(k in text.lower() for k in ["video","footage","tour","demo","walkthrough"]):
+            queries.append({"q": text + " video", "type":"video", "max_results":4})
+        plan = {"id": uid("plan"), "queries": queries, "created_at": now_iso(), "instructions": "Run these queries locally on user's device and return structured results (web/images/videos/audio)."}
+        return plan
+
+    def verify(self, plan_results: Optional[Dict[str,Any]], local_tool_outputs: Optional[List[Dict[str,Any]]]=None) -> Dict[str,Any]:
+        # compute reliability, controversy, citations
+        citations = []
+        reliability = 0.4
+        controversy = 0.0
+        if plan_results:
+            web = plan_results.get("web") or []
+            domains = set()
+            for r in web:
+                u = r.get("url") or r.get("link") or ""
+                domains.add(u.split("/")[2] if "/" in u else u)
+                citations.append({"title": r.get("title"), "url": u, "date": r.get("date")})
+            reliability = min(1.0, 0.2 + 0.1*len(domains))
+            # controversy: if two top results contradict in short text heuristics
+            if len(web) >= 2:
+                s0 = web[0].get("snippet","").lower()
+                s1 = web[1].get("snippet","").lower()
+                if any(w in s1 for w in ["not", "contradict", "dispute"]) or any(w in s0 for w in ["not","contradict","dispute"]):
+                    controversy = 0.5
+        # tool outputs contribution
+        tool_bonus = 0.0
+        for t in (local_tool_outputs or []):
+            if t.get("tool") == "image" and t.get("ok"): tool_bonus += 0.2
+            if t.get("tool") == "math" and not t.get("result","").lower().startswith("math error"): tool_bonus += 0.2
+        confidence = min(1.0, reliability*0.6 + tool_bonus)
+        return {"confidence": confidence, "reliability": reliability, "controversy": controversy, "citations": citations}
+
+    def opportunistic_learning(self, plan_results: Dict[str,Any]):
+        # extract images/videos/audio and store as skill benchmarks if quality high
+        images = plan_results.get("images", []) or []
+        for im in images:
+            path = im.get("path") or im.get("url")
+            # naive quality score
+            score = float(im.get("quality_score", 0.6))
+            tag = im.get("tags",[ "web" ])[0] if im.get("tags") else "web"
+            meta = {"source": path, "desc": im.get("caption") or im.get("alt") or "", "origin":"user_client"}
+            self.csv.add_benchmark(tag, "image", meta, score)
+        videos = plan_results.get("videos", []) or []
+        for v in videos:
+            path = v.get("path") or v.get("url")
+            score = float(v.get("quality_score", 0.6))
+            tag = v.get("tags",[ "web" ])[0] if v.get("tags") else "web"
+            meta = {"source": path, "desc": v.get("caption") or "", "origin":"user_client"}
+            self.csv.add_benchmark(tag, "video", meta, score)
+        audios = plan_results.get("audios", []) or []
+        for a in audios:
+            path = a.get("path") or a.get("url")
+            score = float(a.get("quality_score", 0.6))
+            tag = a.get("tags",[ "web" ])[0] if a.get("tags") else "web"
+            meta = {"source": path, "desc": a.get("caption") or "", "origin":"user_client"}
+            self.csv.add_benchmark(tag, "audio", meta, score)
+        # push to global sync immediately
+        if GLOBAL_SYNC_ENABLED:
+            payload = {"type":"skill_update", "time": now_iso(), "added": len(images)+len(videos)+len(audios)}
+            # best-effort: broadcast via websocket
+            if self.global_sync:
+                loop = asyncio.new_event_loop()
                 try:
-                    arr = reader.get_data(i)
-                    p = tmp_path(f"_frame{i}.jpg")
-                    Image.fromarray(arr).save(p)
-                    frames.append(p)
-                    if len(frames) >= max_frames: break
+                    loop.run_until_complete(self.global_sync.ws_broadcast(payload))
                 except Exception:
-                    continue
-            return {"frames": frames, "count": len(frames)}
-        except Exception as e:
-            return {"error": str(e)}
-
-    def interpolate_fps(self, video_path: str, factor:int=2) -> Dict[str,Any]:
-        """
-        Functional, dependency-light interpolation by frame duplication.
-        This increases frame count (and apparent FPS) without optical flow.
-        Produces a new MP4 if imageio-ffmpeg is available; else returns extracted frames.
-        """
-        try:
-            import imageio, imageio.v3 as iio
-            reader = imageio.get_reader(video_path)
-            fps = reader.get_meta_data().get("fps", 24)
-            new_fps = max(1, int(fps*factor))
-            out_path = tmp_path(".mp4")
+                    pass
+
+    def handle_plan_results(self, plan_id: str, plan_results: Dict[str,Any], local_tool_outputs: Optional[List[Dict[str,Any]]]=None):
+        # verify, opportunistic learn, store facts in VFS
+        v = self.verify(plan_results, local_tool_outputs)
+        # store simple fact example: top web title as a fact
+        web = plan_results.get("web", []) or []
+        if web:
+            top = web[0]
+            claim = top.get("title","").strip()
+            value = top.get("snippet","").strip()
+            sources = [{"url": top.get("url") or top.get("link"), "title": top.get("title"), "date": top.get("date")}]
+            self.vfs.store_fact(claim, value, sources, float(v.get("confidence",0.4)), controversy=v.get("controversy",0.0))
+        # opportunistic learning
+        self.opportunistic_learning(plan_results)
+        return v
+
+# --------------------------
+# Auto-backup & Export
+# --------------------------
+class BackupManager:
+    def __init__(self, db: LocalDB, data_dir: Path, backup_dir: Path):
+        self.db = db
+        self.data_dir = data_dir
+        self.backup_dir = backup_dir
+        self.lock = threading.Lock()
+    def create_backup(self) -> str:
+        with self.lock:
+            ts = now_iso().replace(":", "-")
+            out_path = self.backup_dir / f"mm_backup_{ts}.zip"
+            with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED) as zf:
+                # include SQLite DB file if exists
+                try:
+                    dbf = Path(self.db.path)
+                    if dbf.exists():
+                        zf.write(str(dbf), arcname=dbf.name)
+                except Exception:
+                    pass
+                # include JSON DB mirrors (facts/skills/modules)
+                # try to export via DB list functions
+                try:
+                    facts = self.db.list_facts()
+                    skills = self.db.list_skills()
+                    mods = self.db.list_modules()
+                    zf.writestr("facts.json", json.dumps(facts, ensure_ascii=False, indent=2))
+                    zf.writestr("skills.json", json.dumps(skills, ensure_ascii=False, indent=2))
+                    zf.writestr("modules.json", json.dumps(mods, ensure_ascii=False, indent=2))
+                except Exception:
+                    pass
+            bid = self.db.add_backup(str(out_path))
+            return str(out_path)
+    def download_backup_path(self) -> Optional[str]:
+        # return latest backup path
+        cur = self.db.conn.cursor()
+        cur.execute("SELECT id,path,created_at FROM backups ORDER BY created_at DESC LIMIT 1")
+        r = cur.fetchone()
+        if not r: return None
+        return r[1]
+
+# --------------------------
+# Main Orchestrator (upgrades v6.SuperAgent)
+# --------------------------
+class SuperAgentV7:
+    def __init__(self):
+        self.db_path = DATA_DIR / "multimodular_v7.db"
+        self.db = LocalDB(self.db_path)
+        self.vfs = VFS(self.db)
+        self.csv = CreativeSkillVault(self.db)
+        self.global_sync = GlobalSync(self.db, hf_token=HUGGINGFACE_TOKEN, repo=GLOBAL_SYNC_REPO)
+        # module manager uses GLOBAL_SYNC_SIGNING_PUBKEY if present
+        self.module_mgr = ModuleManager(self.db, signing_pubkey_pem=GLOBAL_SYNC_SIGNING_PUBKEY)
+        self.chb = CHB(self.db, self.vfs, self.csv, self.module_mgr, self.global_sync)
+        self.backup_mgr = BackupManager(self.db, DATA_DIR, BACKUP_DIR)
+        # start realtime ws server in background for incoming updates
+        if websockets is not None:
             try:
-                writer = imageio.get_writer(out_path, fps=new_fps)
-                for frame in reader:
-                    # duplicate frames "factor" times
-                    for _ in range(factor):
-                        writer.append_data(frame)
-                writer.close()
-                return {"status":"ok", "output": out_path, "fps": new_fps}
+                self.global_sync.run_ws_in_thread()
             except Exception as e:
-                return {"status":"partial", "reason": f"writer failed: {e}"}
-        except Exception as e:
-            return {"status":"error", "reason": str(e)}
-
-# ---------------------------
-# Voice engine (register, embed, TTS)
-# ---------------------------
-class VoiceEngine:
-    def __init__(self, provenance: ProvenanceManager, safety: SafetyManager, piper_bin=None, piper_voice=None):
-        self.prov = provenance
-        self.safety = safety
-        self.piper_bin = piper_bin
-        self.piper_voice = piper_voice
-        self.profiles: Dict[str,Dict[str,Any]] = {}
-
-    def extract_embedding(self, wav_path: str) -> Optional[List[float]]:
+                log.warning("ws server thread failed: %s", e)
+    # --- user-driven search plan creation (CHB.plan)
+    def plan_search(self, text: str) -> Dict[str,Any]:
+        return self.chb.plan(text)
+    # --- client submits results
+    def submit_plan_results(self, plan_id: str, results: Dict[str,Any]) -> Dict[str,Any]:
+        # Accept results; pass to CHB for verify & learning
+        v = self.chb.handle_plan_results(plan_id, results)
+        # create backup after major update
         try:
-            with open(wav_path,"rb") as f:
-                b = f.read()
-            h = hashlib.sha256(b).digest()
-            return [float(x)/255.0 for x in h[:192]]
+            bp = self.backup_mgr.create_backup()
+            log.info("backup created: %s", bp)
         except Exception:
-            return None
-
-    def register(self, user_id: str, wav_path: str, consent_text: str, block_public=True):
-        token = self.safety.record_consent(user_id, consent_text)
-        emb = self.extract_embedding(wav_path)
-        if block_public and self.safety.check_public_figure(emb):
-            return {"status":"rejected","reason":"protected_speaker"}
-        vid = f"voice_{hashlib.sha1((user_id+str(time.time())).encode()).hexdigest()[:10]}"
-        self.profiles[vid] = {"user_id": user_id, "embedding": emb, "consent": token}
-        return {"status":"ok", "voice_id": vid}
-
-    def synthesize(self, voice_id:Optional[str], text:str, emotion:Optional[str]=None, rate:float=1.0, fmt:str="wav"):
-        if not self.safety.is_allowed(text):
-            return None
-        out = tmp_path(f".{fmt}")
-        # Prefer Piper if configured (local CLI)
-        if self.piper_bin and self.piper_voice and os.path.exists(self.piper_bin) and os.path.exists(self.piper_voice):
-            try:
-                p = subprocess.Popen([self.piper_bin, "-m", self.piper_voice, "-f", out],
-                                     stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
-                p.communicate(input=text, timeout=60)
-                if os.path.exists(out):
-                    with open(out,"rb") as f:
-                        b = f.read()
-                    b2 = self.prov.watermark_audio(b)
-                    with open(out,"wb") as f: f.write(b2)
-                    self.prov.attach(out, Provenance(model="piper", version="1.0",
-                                                     consent=self.profiles.get(voice_id, {}).get("consent")))
-                    return out
-            except Exception as e:
-                logger.debug("piper fail: %s", e)
-        # Fallback pyttsx3
-        try:
-            import pyttsx3
-            eng = pyttsx3.init()
-            try: eng.setProperty("rate", int(200*rate))
-            except Exception: pass
-            eng.save_to_file(text, out); eng.runAndWait()
-            with open(out,"rb") as f:
-                b = f.read()
-            b2 = self.prov.watermark_audio(b)
-            with open(out,"wb") as f: f.write(b2)
-            self.prov.attach(out, Provenance(model="pyttsx3", version="1.0"))
-            return out
-        except Exception as e:
-            logger.debug("pyttsx3 fail: %s", e)
-            # fallback: silent wav (keeps pipeline functional)
+            log.exception("backup failed")
+        # push to global sync (best-effort)
+        if GLOBAL_SYNC_ENABLED:
             try:
-                import wave, struct
-                fr=16000; dur=max(1, min(5, int(len(text)/10)))
-                with wave.open(out,'w') as wf:
-                    wf.setnchannels(1); wf.setsampwidth(2); wf.setframerate(fr)
-                    for _ in range(fr*dur): wf.writeframes(struct.pack('<h',0))
-                self.prov.attach(out, Provenance(model="silence", version="0.1"))
-                return out
+                payload = {"type":"fact_skill_update", "ts": now_iso(), "payload": {"summary":"update","plan_id": plan_id}}
+                # attempt HF push or WS broadcast
+                # HF push requires implementation using huggingface_hub; we leave a placeholder
+                asyncio.run(self.global_sync.ws_broadcast(payload))
             except Exception:
-                return None
-
-# ---------------------------
-# MathEngine (SymPy)
-# ---------------------------
-class MathEngine:
-    def __init__(self):
-        try:
-            import sympy as sp
-            self.sp = sp
-        except Exception:
-            self.sp = None
-
-    def solve(self, expr: str) -> str:
-        if not self.sp: return "[sympy missing]"
-        try:
-            parsed = self.sp.sympify(expr)
-            return str(self.sp.simplify(parsed))
-        except Exception as e:
-            return f"[math-error] {e}"
-
-    def eval_numeric(self, expr: str) -> str:
-        if not self.sp: return "[sympy missing]"
-        try:
-            return str(self.sp.N(self.sp.sympify(expr)))
-        except Exception as e:
-            return f"[math-error] {e}"
-
-    def safe_eval(self, text: str) -> Optional[str]:
-        # detect simple math patterns like "2+2", "sqrt(2)", etc.
-        if re.fullmatch(r"[0-9\.\+\-\*\/\(\)\s^sqrtpiE]+", text.replace("**","^")):
-            return self.eval_numeric(text)
-        return None
-
-# ---------------------------
-# Code Sandbox (lightweight)
-# ---------------------------
-class CodeSandbox:
-    def __init__(self):
-        pass
-    def run(self, code:str, timeout: int=5) -> Dict[str,Any]:
-        # VERY light sandbox; for real isolation use subprocess + rlimits in your env.
-        try:
-            import io, contextlib
-            buf = io.StringIO()
-            ns = {}
-            with contextlib.redirect_stdout(buf):
-                exec(code, {"__builtins__": {"print": print, "range": range, "len": len}}, ns)
-            out = buf.getvalue()
-            return {"stdout": out, "keys": list(ns.keys())}
-        except Exception as e:
-            return {"error": str(e)}
-
-# ---------------------------
-# Agents
-# ---------------------------
-class Agent:
-    def __init__(self, name:str, model: LocalLLM, tools:Dict[str,Any]=None):
-        self.name = name
-        self.model = model
-        self.tools = tools or {}
-
-    def act(self, prompt:str, max_tokens=256, temperature=0.7):
-        try:
-            if not self.model.ready():
-                self.model.load()
-            return self.model.generate(prompt, max_tokens=max_tokens, temperature=temperature)
-        except Exception as e:
-            logger.error("Agent %s failed: %s", self.name, e)
-            return f"[{self.name}-error] {e}"
-
-class AgentHub:
-    def __init__(self, llm: LocalLLM, img_engine: ImageEngine, vid_engine: VideoEngine):
-        self.research = Agent("ResearchAgent", llm, {"web": None})
-        self.coder = Agent("CoderAgent", llm, {"exec": CodeSandbox()})
-        self.designer = Agent("DesignerAgent", llm, {"image": img_engine})
-        self.vid = Agent("VideoAgent", llm, {"video": vid_engine})
-
-    def coordinate(self, user_request:str):
-        if any(k in user_request.lower() for k in ["code","implement","script"]):
-            return self.coder.act(f"Write code for: {user_request}", max_tokens=512)
-        if any(k in user_request.lower() for k in ["design","image","generate"]):
-            return self.designer.act(f"Create an image plan: {user_request}", max_tokens=256)
-        return self.research.act(f"Research and summarize: {user_request}", max_tokens=512)
-
-# ---------------------------
-# Perception (lightweight multimodal glue)
-# ---------------------------
-class Perception:
-    EMOJI_MAP = {
-        "😂":"joy","😅":"relief","😭":"sad","😡":"anger","😍":"affection","😎":"confident",
-        "🤔":"thinking","😴":"tired","🙄":"annoyed","😇":"polite","😁":"happy","🤣":"joy"
-    }
-    def text_emotion_tags(self, text:str) -> List[str]:
-        tags=set()
-        for ch in text:
-            if ch in self.EMOJI_MAP: tags.add(self.EMOJI_MAP[ch])
-        if re.search(r"\b(sad|upset|angry|frustrated)\b", text.lower()): tags.add("negative")
-        if re.search(r"\b(happy|great|awesome|love)\b", text.lower()): tags.add("positive")
-        return sorted(tags)
-
-# ---------------------------
-# Confidence Scorer
-# ---------------------------
-class ConfidenceScorer:
-    def score(self, *,
-              source_reliability: float,
-              recency: float,
-              agreement: float,
-              self_consistency: float,
-              retrieval_strength: float,
-              contradictions: float) -> float:
-        # conservative weighting; each in [0,1]
-        w1,w2,w3,w4,w5,w6 = 0.22,0.18,0.18,0.18,0.14,0.10
-        s = (w1*source_reliability +
-             w2*recency +
-             w3*agreement +
-             w4*self_consistency +
-             w5*retrieval_strength -
-             w6*contradictions)
-        return max(0.0, min(1.0, s))
-
-# ---------------------------
-# Close-to-Human Brain (CHB)
-# ---------------------------
-class CloseToHumanBrain:
-    def __init__(self, llm: LocalLLM, longmem: LongTermMemory, kg: KnowledgeGraph,
-                 vfs: VersionedFactStore, web: WebSearch, img: ImageEngine):
-        self.llm = llm
-        self.long = longmem
-        self.kg = kg
-        self.vfs = vfs
-        self.web = web
-        self.img = img
-        self.math = MathEngine()
-        self.perc = Perception()
-        self.scorer = ConfidenceScorer()
-
-    # ---- Retrieval
-    def retrieve(self, user_text: str, k:int=5) -> Tuple[str, List[Dict[str,str]]]:
-        mem_hits = self.long.search(user_text, top_k=min(8,k))
-        mem_ctx = "\n".join([h["text"] for h in mem_hits]) if mem_hits else ""
-        web_hits = self.web.search(user_text, max_results=3) if self.web.enabled else []
-        return mem_ctx, web_hits
-
-    # ---- Drafting
-    def multi_draft(self, prompt_base: str, drafts:int=3, max_tokens:int=384) -> List[str]:
-        outs=[]
-        temps=[0.4, 0.7, 1.0][:max(1,drafts)]
-        if not self.llm.ready():
-            self.llm.load()
-        for t in temps:
-            out = self.llm.generate(prompt_base, max_tokens=max_tokens, temperature=t)
-            outs.append(out)
-        return outs
-
-    # ---- Verification helpers
-    def _estimate_reliability(self, sources: List[Dict[str,Any]]) -> float:
-        if not sources: return 0.4
-        rel=0.0
-        for s in sources:
-            t = s.get("type","")
-            if t=="memory": rel += 0.6
-            elif t=="kg": rel += 0.7
-            elif t=="web": rel += 0.65
-            elif t in ("vision","audio","video"): rel += 0.55
-            else: rel += 0.5
-        return min(1.0, rel / max(1,len(sources)))
-
-    def _recency(self, sources: List[Dict[str,Any]]) -> float:
-        if not sources: return 0.3
-        ages=[]
-        now=now_ts()
-        for s in sources:
-            ts = s.get("time") or now
-            ages.append(max(0.0, now - ts))
-        avg = sum(ages)/len(ages)
-        # map age (in seconds) to [0,1] with simple decay (~1 day half-life)
-        day = 86400.0
-        return max(0.0, min(1.0, 1.0/(1.0 + (avg/day))))
-
-    def _agreement(self, claims: List[str]) -> float:
-        # crude token overlap agreement
-        if not claims: return 0.0
-        base=set(re.findall(r"\w+", claims[0].lower()))
-        agree=1
-        for c in claims[1:]:
-            toks=set(re.findall(r"\w+", c.lower()))
-            if len(base & toks) > 0: agree += 1
-        return agree/len(claims)
-
-    def _self_consistency(self, drafts: List[str]) -> float:
-        # measure average pairwise Jaccard of word sets
-        if not drafts: return 0.0
-        sets=[set(re.findall(r"\w+", d.lower())) for d in drafts]
-        if len(sets)==1: return 1.0
-        pair_scores=[]
-        for i in range(len(sets)):
-            for j in range(i+1,len(sets)):
-                a,b=sets[i],sets[j]
-                inter=len(a & b); union=len(a | b) or 1
-                pair_scores.append(inter/union)
-        return sum(pair_scores)/len(pair_scores)
-
-    def _retrieval_strength(self, mem_ctx: str) -> float:
-        if not mem_ctx: return 0.4
-        # simple function of context length
-        L = len(mem_ctx.split())
-        return max(0.4, min(1.0, math.log10(1+L)/2))
-
-    # ---- Verify & Synthesize
-    def verify_and_respond(self, user_id:str, user_text: str,
-                           preferred_lang: Optional[str]=None) -> Dict[str,Any]:
-        # Perception
-        emotion_tags = self.perc.text_emotion_tags(user_text)
-
-        # Retrieve evidence
-        mem_ctx, web_hits = self.retrieve(user_text, k=6)
-        web_ctx = "\n".join([f"{h['title']} ({h['link']}) — {h.get('snippet','')}" for h in web_hits]) if web_hits else ""
-        citations = [{"type":"web","ref": h["link"], "title": h["title"], "time": now_ts()} for h in web_hits]
-
-        # Construct base prompt
-        prompt = (
-            "Persona: helpful, precise assistant.\n"
-            "Use given memory and web snippets as *evidence*.\n"
-            "If a claim is uncertain, narrow it or state limits.\n\n"
-            f"Memory Evidence:\n{mem_ctx}\n\nWeb Evidence:\n{web_ctx}\n\n"
-            f"User: {user_text}\nAssistant:"
-        )
-
-        # Multi-draft
-        drafts = self.multi_draft(prompt, drafts=3, max_tokens=512)
-
-        # Self-consistency + basic math check (if any plain expression present)
-        math_value = None
-        m = re.search(r"(?:calculate|solve)\s*([0-9\.\+\-\*\/\(\)\s^sqrtpiE]+)", user_text, re.I)
-        if m:
-            math_value = self.math.safe_eval(m.group(1))
-
-        # Build sources list (memory + web)
-        sources = citations[:]
-        if mem_ctx:
-            sources.append({"type":"memory","ref":"longterm_memory","title":"long-term memory","time": now_ts()})
-
-        # Score
-        source_rel = self._estimate_reliability(sources)
-        recency = self._recency(sources)
-        agree = self._agreement(drafts)
-        self_cons = self._self_consistency(drafts)
-        retr = self._retrieval_strength(mem_ctx)
-        contradictions = 0.0
-
-        # Compare with VFS (detect contradictions for simple exact match claims)
-        # We extract simple "X is Y" patterns from drafts; very lightweight.
-        simple_claims=[]
-        for d in drafts:
-            for sent in re.split(r"[.\n]", d):
-                m2 = re.search(r"^([\w\s\-]{3,})\s+is\s+([\w\s\-\%\.]{2,})$", sent.strip(), re.I)
-                if m2:
-                    c = f"{m2.group(1).strip()} is"
-                    v = m2.group(2).strip()
-                    simple_claims.append((c,v))
-        # check against latest
-        for c,v in simple_claims:
-            latest = self.vfs.latest(c)
-            if latest and latest.get("value") and latest["value"].strip().lower()!=v.lower():
-                contradictions += 0.5  # penalize disagreement with stored fact
-
-        conf = self.scorer.score(
-            source_reliability=source_rel,
-            recency=recency,
-            agreement=agree,
-            self_consistency=self_cons,
-            retrieval_strength=retr,
-            contradictions=contradictions
-        )
-
-        # Choose the most concise draft
-        best = min(drafts, key=lambda s: len(s) if s else 1e9)
-
-        # If math was requested & computed, splice it in with highest certainty
-        if math_value and "[math-error]" not in math_value and "sympy missing" not in math_value:
-            best = f"{best}\n\nMath check: {math_value}"
-
-        # If confidence < threshold, trim to certain subset
-        min_conf = MODEL_CONFIG["chb_min_confidence"]
-        if conf < min_conf:
-            # Provide narrowed/certain answer: we extract sentences with highest overlap across drafts.
-            sent_scores=[]
-            sents = [s.strip() for s in re.split(r"(?<=[\.\!\?])\s+", best) if s.strip()]
-            for s in sents:
-                count=sum(1 for d in drafts if s.lower() in d.lower())
-                sent_scores.append((count, s))
-            sent_scores.sort(reverse=True)
-            certain = " ".join([s for cnt,s in sent_scores if cnt>=2])  # present in >=2 drafts
-            if not certain:
-                certain = "I'm not fully confident. Here's what is most certain from the evidence I have."
-            best = certain
-
-        # Record any simple claims to VFS as new knowledge (with provenance)
-        for c,v in simple_claims[:3]:
-            self.vfs.add_or_update(claim=c, value=v, sources=sources, confidence=float(conf))
-
-        # Build final message with optional citations
-        if citations:
-            cites = "\n".join([f"- {c['title']} — {c['ref']}" for c in citations])
-            best_out = f"{best}\n\nConfidence: {conf:.2f}\nSources:\n{cites}"
-        else:
-            best_out = f"{best}\n\nConfidence: {conf:.2f}"
-
-        return {"reply": best_out, "confidence": conf, "citations": citations}
-
-# ---------------------------
-# Orchestrator (ties everything with CHB)
-# ---------------------------
-class SuperAgent:
-    def __init__(self, config:dict):
-        self.cfg = config
-        self.prov = ProvenanceManager()
-        self.safety = SafetyManager(blocklist=config.get("safety_blocklist"))
-        self.short = ShortTermMemory()
-        self.long = LongTermMemory(index_dir=config.get("faiss_index_dir"), embed_model_name=config.get("embedder"))
-        self.kg = KnowledgeGraph(config.get("knowledge_graph_path"))
-        self.vfs = VersionedFactStore(config.get("vfs_file"))
-        self.web = WebSearch(enabled=bool(config.get("allow_web_search")), cache_file=config.get("web_cache_file"))
-        # LLM
-        self.llm = LocalLLM(model_path=config.get("llm",{}).get("model_path"),
-                            backend=config.get("llm",{}).get("backend","transformers"),
-                            device=config.get("device","auto"),
-                            quantize=config.get("llm",{}).get("quantize",None))
-        # image & video engines
-        self.image = ImageEngine(base=config.get("sdxl_base"), refiner=config.get("sdxl_refiner"),
-                                 inpaint=config.get("sdxl_inpaint"), blip=config.get("blip_caption"),
-                                 device=config.get("device","auto"))
-        self.video = VideoEngine()
-        self.voice = VoiceEngine(self.prov, self.safety, piper_bin=config.get("piper_binary"), piper_voice=config.get("piper_voice"))
-        # agents
-        self.agents = AgentHub(self.llm, self.image, self.video)
-        # Close-to-Human Brain
-        self.chb = CloseToHumanBrain(self.llm, self.long, self.kg, self.vfs, self.web, self.image)
-
-    def detect_intent(self, text:str) -> str:
-        t = (text or "").lower().strip()
-        if t.startswith("/img ") or t.startswith("/image "): return "image"
-        if t.startswith("/inpaint "): return "inpaint"
-        if t.startswith("/tts "): return "tts"
-        if t.startswith("/video "): return "video"
-        if t.startswith("/vidinterp "): return "vidinterp"
-        if t.startswith("/kg "): return "kg"
-        if t.startswith("/agent "): return "agent"
-        if any(k in t for k in ["solve", "calculate", "integrate", "differentiate"]): return "math"
-        return "chat"
-
-    def handle(self, user_id:str, text:str, preferred_lang:Optional[str]=None) -> Dict[str,Any]:
-        if not self.safety.is_allowed(text):
-            return {"status":"blocked","reason":"policy"}
-        self.short.push(user_id, "user", text)
+                pass
+        return {"ok": True, "verify": v}
+    # --- expose facts/skills
+    def search_facts(self, q: str):
+        return self.vfs.query(q)
+    def top_skills(self, tag: str, k: int=5):
+        return self.csv.top_by_tag(tag, k)
+    def add_module(self, name: str, code: str, signature_b64: Optional[str]=None, autointegrate: bool=True):
+        return self.module_mgr.integrate_module(name, code, signature_b64, autointegrate)
+    def download_latest_backup(self) -> Optional[str]:
+        return self.backup_mgr.download_backup_path()
+    # direct chat route that funnels through CHB
+    def chat(self, text: str, plan_results: Optional[Dict[str,Any]]=None):
+        # For direct chat, if user provided plan_results (client retrieval), handle them
+        if plan_results:
+            v = self.chb.handle_plan_results(uid("plan"), plan_results)
+            return {"status":"ok", "verify": v}
+        # Simple path: CHB will create plan if needed and answer (we re-use simple plan + verify)
+        plan = self.chb.plan(text)
+        # no client retrieval performed: CHB can still reply with local knowledge (fallback)
+        # For v7 we simply return plan and ask client to run it OR CHB will attempt local infer (fallback)
+        return {"status":"ok", "plan": plan, "hint": "Run the plan on client and submit results via submit_plan_results"}
+
+# --------------------------
+# FastAPI endpoints (added/extended)
+# --------------------------
+if FASTAPI_AVAILABLE:
+    app = FastAPI(title="Multimodular SuperAgent v7.0")
+    AGENT = SuperAgentV7()
+
+    class PlanIn(BaseModel):
+        text: str
 
-        intent = self.detect_intent(text)
-        reply = ""
-        payload: Dict[str,Any] = {}
+    @app.post("/v1/plan_search")
+    async def api_plan_search(inp: PlanIn):
+        plan = AGENT.plan_search(inp.text)
+        return {"ok": True, "plan": plan}
 
+    @app.post("/v1/submit_results")
+    async def api_submit_results(plan_id: str = Form(...), results: str = Form(...)):
         try:
-            if intent == "math":
-                me = self.chb.math
-                expr = re.sub(r"^(solve|calculate)\s*","", text, flags=re.I).strip()
-                res = {"exact": me.solve(expr), "numeric": me.eval_numeric(expr)}
-                reply = json.dumps(res, ensure_ascii=False, indent=2)
-
-            elif intent == "image":
-                prompt = text.split(" ",1)[1] if " " in text else text
-                path = self.image.generate(prompt=prompt)
-                reply = f"[image] {path}"
-
-            elif intent == "inpaint":
-                reply = "Use /v1/image/inpaint API with 'image' and optional 'mask' files."
-
-            elif intent == "tts":
-                content = text.split(" ",1)[1] if " " in text else text
-                path = self.voice.synthesize(None, content)
-                reply = f"[tts] {path}"
-
-            elif intent == "video":
-                reply = "Upload video via /v1/video/analyze to extract keyframes."
-
-            elif intent == "vidinterp":
-                parts = text.split(" ",2)
-                factor = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 2
-                reply = f"Use /v1/video/interpolate (factor={factor}) with a video file."
-
-            elif intent == "kg":
-                reply = "Use /v1/kg endpoints (not exposed in this single file demo) or extend as needed."
-
-            elif intent == "agent":
-                task = text.split(" ",1)[1] if " " in text else ""
-                out = self.agents.coordinate(task)
-                # CHB is the final boss: pass agent output through CHB to verify/format
-                verified = self.chb.verify_and_respond(user_id, f"{task}\n\nAgentDraft:\n{out}", preferred_lang)
-                reply = verified["reply"]
-                payload.update({"confidence": verified["confidence"]})
-
-            else:
-                # Default conversational path goes through CHB (final arbiter)
-                verified = self.chb.verify_and_respond(user_id, text, preferred_lang)
-                reply = verified["reply"]
-                payload.update({"confidence": verified["confidence"]})
-
-        except Exception as e:
-            logger.error("Handle failed: %s", e)
-            reply = f"[error] {e}"
-
-        # Learning: store turn + (optionally) extracted facts handled inside CHB
-        if MODEL_CONFIG.get("auto_learn"):
-            try:
-                self.long.add(user_id, text, kind="turn")
-            except Exception as e:
-                logger.debug("long-term add failed: %s", e)
-
-        self.short.push(user_id, "assistant", reply)
-        return {"status":"ok", "reply": reply, **payload}
-
-# ---------------------------
-# FastAPI + Gradio integration
-# ---------------------------
-try:
-    from fastapi import FastAPI, UploadFile, File, Form
-    from fastapi.responses import FileResponse, JSONResponse
-    from pydantic import BaseModel
-    _FASTAPI = True
-except Exception:
-    _FASTAPI = False
-
-try:
-    import gradio as gr
-    _GRADIO = True
-except Exception:
-    _GRADIO = False
-
-app = FastAPI(title="Multimodal SuperAgent") if _FASTAPI else None
-_AGENT_SINGLETON: Optional[SuperAgent] = None
+            payload = json.loads(results)
+        except Exception:
+            return JSONResponse({"ok": False, "error": "invalid_json"}, status_code=400)
+        out = AGENT.submit_plan_results(plan_id, payload)
+        return out
+
+    @app.post("/v1/facts/search")
+    async def api_facts_search(q: str = Form(...)):
+        res = AGENT.search_facts(q)
+        return {"ok": True, "results": res}
+
+    @app.post("/v1/skills/top")
+    async def api_skills_top(tag: str = Form(...), k: int = Form(5)):
+        res = AGENT.top_skills(tag, k)
+        return {"ok": True, "results": res}
+
+    @app.post("/v1/module/add")
+    async def api_module_add(name: str = Form(...), code: str = Form(...), signature_b64: Optional[str] = Form(None)):
+        out = AGENT.add_module(name, code, signature_b64)
+        return out
+
+    @app.get("/v1/backup/download")
+    async def api_backup_download():
+        p = AGENT.download_latest_backup()
+        if not p or not os.path.exists(p):
+            return JSONResponse({"ok": False, "error": "no_backup"}, status_code=404)
+        return FileResponse(p, media_type="application/zip", filename=os.path.basename(p))
 
-def get_agent() -> SuperAgent:
-    global _AGENT_SINGLETON
-    if _AGENT_SINGLETON is None:
-        # merge env config file if present
-        cfg_path = os.environ.get("SUPERAGENT_CONFIG")
-        cfg = MODEL_CONFIG.copy()
-        if cfg_path and os.path.exists(cfg_path):
+    @app.post("/v1/chat")
+    async def api_chat(text: str = Form(...), plan_results: Optional[str] = Form(None)):
+        if plan_results:
             try:
-                import yaml
-                with open(cfg_path,"r",encoding="utf-8") as f:
-                    y = yaml.safe_load(f) or {}
-                    cfg.update(y)
+                pr = json.loads(plan_results)
             except Exception:
-                pass
-        _AGENT_SINGLETON = SuperAgent(cfg)
-    return _AGENT_SINGLETON
-
-if _FASTAPI and app is not None:
-    class ChatIn(BaseModel):
-        user_id: str
-        text: str
-        preferred_lang: Optional[str]=None
-
-    @app.post("/v1/chat")
-    async def api_chat(inp: ChatIn):
-        return get_agent().handle(inp.user_id, inp.text, inp.preferred_lang)
-
-    @app.post("/v1/register_voice")
-    async def api_reg_voice(user_id: str = Form(...), consent: str = Form(...), file: UploadFile = File(...)):
-        tmpf = tmp_path("_voice.wav")
-        with open(tmpf,"wb") as f: f.write(await file.read())
-        return get_agent().voice.register(user_id, tmpf, consent)
-
-    @app.post("/v1/tts")
-    async def api_tts(voice_id: Optional[str]=Form(None), text: str = Form(...), fmt: str = Form("wav")):
-        p = get_agent().voice.synthesize(voice_id, text, fmt=fmt)
-        if p and os.path.exists(p): return FileResponse(p, media_type="audio/wav")
-        return JSONResponse({"error":"tts_failed"})
-
-    @app.post("/v1/image/generate")
-    async def api_img_gen(prompt: str = Form(...)):
-        p = get_agent().image.generate(prompt)
-        if p and os.path.exists(p): return FileResponse(p, media_type="image/png")
-        return JSONResponse({"error":"image_unavailable"})
-
-    @app.post("/v1/image/inpaint")
-    async def api_img_inpaint(prompt: str = Form(...), image: UploadFile = File(...), mask: UploadFile = File(None)):
-        img_tmp = tmp_path("_img")
-        with open(img_tmp,"wb") as f: f.write(await image.read())
-        mask_tmp=None
-        if mask:
-            mask_tmp = tmp_path("_mask"); open(mask_tmp,"wb").write(await mask.read())
-        p = get_agent().image.inpaint(img_tmp, mask_tmp, prompt)
-        if p and os.path.exists(p): return FileResponse(p, media_type="image/png")
-        return JSONResponse({"error":"inpaint_failed"})
-
-    @app.post("/v1/video/analyze")
-    async def api_vid(file: UploadFile = File(...)):
-        tmpf = tmp_path("_vid")
-        with open(tmpf,"wb") as f: f.write(await file.read())
-        return get_agent().video.analyze(tmpf)
-
-    @app.post("/v1/video/interpolate")
-    async def api_vid_interp(factor: int = Form(2), file: UploadFile = File(...)):
-        tmpf = tmp_path("_vid")
-        with open(tmpf,"wb") as f: f.write(await file.read())
-        return get_agent().video.interpolate_fps(tmpf, factor=max(2, int(factor)))
-
-    @app.post("/v1/memory/export")
-    async def mem_export():
-        return get_agent().long.export_all()
-
-    @app.post("/v1/memory/import")
-    async def mem_import(items: List[Dict[str,Any]]):
-        get_agent().long.import_bulk(items)
-        return {"status":"ok","count":len(items)}
-
-    @app.post("/v1/web/toggle")
-    async def web_toggle(enabled: bool = Form(...)):
-        get_agent().web.enabled = bool(enabled); return {"enabled":get_agent().web.enabled}
-
-# ---------------------------
-# Optional Gradio demo (runs when module executed)
-# ---------------------------
-def launch_gradio():
-    if not _GRADIO:
-        logger.warning("Gradio not installed")
-        return
-    agent = get_agent()
-    with gr.Blocks(title="Multimodal SuperAgent") as demo:
-        gr.Markdown("# Multimodal SuperAgent v6.0 (CHB)")
-        with gr.Row():
-            user_id = gr.Textbox(value="user1", label="User ID")
-            prompt = gr.Textbox(label="Prompt")
-            btn = gr.Button("Send")
-            out = gr.Markdown(label="Reply")
-            def send(u,p):
-                res = agent.handle(u,p)
-                return res.get("reply","")
-            btn.click(send, [user_id,prompt], out)
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+                return JSONResponse({"ok": False, "error": "invalid_plan_results"}, status_code=400)
+            out = AGENT.chat(text, pr)
+        else:
+            out = AGENT.chat(text, None)
+        return out
+else:
+    app = None
+    AGENT = SuperAgentV7()
+
+# --------------------------
+# CLI demo & execution
+# --------------------------
+def demo_run():
+    print("Multimodular SuperAgent v7.0 - Demo")
+    a = AGENT
+    # 1) plan search
+    plan = a.plan_search("futuristic electric motorcycle neon blue lights battery tech 2025")
+    print("Plan:", plan)
+    # Simulate client retrieval
+    sample_results = {
+        "web": [
+            {"title":"Solid-state battery prototype reaches 500 Wh/kg", "url":"https://example.org/article", "snippet":"Researchers at X report 500 Wh/kg...", "date": now_iso()}
+        ],
+        "images":[
+            {"path": str(BASE_DIR / "demo_motorcycle.png"), "quality_score": 0.92, "caption":"Futuristic motorcycle concept", "tags":["motorcycle","futuristic"]}
+        ],
+        "videos": [],
+        "audios": []
+    }
+    # ensure demo image exists
+    try:
+        from PIL import Image, ImageDraw
+        img = Image.new("RGB", (640,480), (20,20,30))
+        d = ImageDraw.Draw(img); d.text((20,20), "Demo motorcycle", fill=(200,200,255))
+        img.save(BASE_DIR / "demo_motorcycle.png")
+    except Exception:
+        pass
+    res = a.submit_plan_results(plan["id"], sample_results)
+    print("Submit results ->", res)
+    # show facts
+    facts = a.search_facts("solid-state")
+    print("Facts:", facts)
+    # download backup
+    path = a.download_latest_backup()
+    print("Backup path:", path)
 
-# ---------------------------
-# CLI
-# ---------------------------
 if __name__ == "__main__":
     import argparse
     ap = argparse.ArgumentParser()
     ap.add_argument("--demo", action="store_true")
-    ap.add_argument("--gradio", action="store_true")
+    ap.add_argument("--runserver", action="store_true")
+    ap.add_argument("--port", type=int, default=8000)
     args = ap.parse_args()
     if args.demo:
-        a = get_agent()
-        print("Chat:", a.handle("user1","Hello 😁, calculate 2*(3+4).")["reply"])
-        print("Math:", a.handle("user1","solve 2*x+1=5")["reply"])
-        print("Interp hint:", a.handle("user1","/vidinterp 2")["reply"])
-    if args.gradio:
-        launch_gradio()
+        demo_run()
+    elif args.runserver and FASTAPI_AVAILABLE:
+        import uvicorn
+        uvicorn.run("multimodular_modul version 7.0:app", host="0.0.0.0", port=args.port, reload=False)
+    else:
+        print("Run with --demo or --runserver. FASTAPI available:", FASTAPI_AVAILABLE)