app.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# HIVE 🐝 FULL MERGED ALL-IN-ONE **OPTIMIZED**
# Offline-first + Online updates + Auto Wi-Fi + RBAC + Multilingual Voice (ASR/TTS + Phonics) 
# + Internal Optimization Stack (Change Manager: propose ➡️ sandbox ➡️ A/B test ➡️ apply/rollback with Owner policy) 
# Upload this single file and requirements.txt to a Hugging Face Space (or run locally).
#  - python app.py

# --- BEGIN MEMORY MANIFEST (auto-updated) ---
# (This block is auto-written by Hive to record what datasets/files
#  have already been converted into memory (curves). Do not edit by hand.)
MEMORY_MANIFEST = {
    "updated_ts": 0,
    "datasets_done": [],
    "vectors_total": 0,
    "notes": "Set HIVE_ALLOW_SELF_WRITE_MANIFEST=0 to stop auto-updates."
}
# --- END MEMORY MANIFEST ---


import os, sys, re, json, time, shutil, tempfile, subprocess, platform, socket, threading, importlib, hashlib, unicodedata, urllib.request, base64, random
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Tuple
from pathlib import Path as _Path

# IMPORTANT: Import FAISS first to avoid segmentation faults on some systems.
# This is a known issue where FAISS needs to be imported before other libraries
# like numpy or torch that might use conflicting low-level libraries.
# ----------- light bootstrap (safe) -----------
def _ensure(pkgs: List[str]):
    for p in pkgs:  # type: ignore
        try:
            subprocess.check_call([sys.executable, "-m", "pip", "install", "--upgrade", p], stdout=sys.stdout, stderr=sys.stderr)
        except Exception:
            print(f"Could not install {p}. Please check the output above for details.")

try:
    import faiss
except (ImportError, ModuleNotFoundError):
    _ensure(["faiss-cpu>=1.8.0"])
    import faiss

_ensure(["numpy>=1.24.0", "psutil==5.9.8", "requests>=2.31.0", "gradio>=4.44.0", "sentence-transformers>=3.0.0", "faiss-cpu>=1.8.0",
         "transformers>=4.44.0", "accelerate>=0.33.0", "datasets>=2.21.0", "soundfile>=0.12.1", "faster-whisper>=1.0.0", "langid>=1.1.6", "webrtcvad>=2.0.10",
         "huggingface-hub>=0.23.0,<1.0", "piper-tts>=1.2.0", "g2p_en>=2.1.0", "librosa>=0.10.1", "scikit-learn>=1.1.0", "feedparser>=6.0.11", "duckduckgo-search>=6.2.10",
         "keyring>=24.3.1"])
import collections, logging
import numpy as np, psutil, requests, feedparser, langid, librosa, gradio as gr, soundfile as sf, struct, queue
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
from faster_whisper import WhisperModel
from piper.voice import PiperVoice
from duckduckgo_search import DDGS
from g2p_en import G2p
from sklearn.metrics.pairwise import cosine_similarity
from concurrent.futures import ThreadPoolExecutor

# --- Setup Logging ---
logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] [%(levelname)s] [%(threadName)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    stream=sys.stdout,
    force=True
)

try:
    import pvporcupine
    _HAVE_PVP=True
except ImportError:
    _HAVE_PVP=False

try:
    import webrtcvad
    _HAVE_VAD=True
except ImportError:
    _HAVE_VAD=False

try:
    import torch
except Exception:
    torch=None

from transformers import StoppingCriteria, StoppingCriteriaList, TextIteratorStreamer

class StopOnTokens(StoppingCriteria):
    def __init__(self, stop_token_ids: List[int]):
        self.stop_token_ids = stop_token_ids

    def __call__(self, input_ids: "torch.LongTensor", scores: "torch.FloatTensor", **kwargs) -> bool:
        for stop_id in self.stop_token_ids:
            if input_ids[0][-1] == stop_id:
                return True
        return False

try:
    import faiss
except Exception:
    subprocess.check_call([sys.executable,"-m","pip","install","--upgrade","faiss-cpu>=1.8.0"])
    import faiss

# Optional vision
try:
    import cv2; _HAVE_CV=True
except Exception:
    _HAVE_CV=False
try:
    from PIL import Image
    import pytesseract; _HAVE_TESS=True and _HAVE_CV
except Exception:
    _HAVE_TESS=False

try:
    import keyring
except Exception:
    keyring=None

# ----------------------- config -----------------------
def ENV(name, default=None, cast=str):
    v=os.getenv(name, default)
    if v is None: return None
    if cast is bool: return str(v).lower() in ("1","true","yes","on")
    if cast is int:
        try: return int(v) # type: ignore
        except (ValueError, TypeError): return int(float(v))
    return v

CFG={
    # auto-archive memory to curves.tar.gz
    "HIVE_AUTO_ARCHIVE": ENV("HIVE_AUTO_ARCHIVE", "1", bool),
    "HIVE_AUTO_ARCHIVE_MODE": ENV("HIVE_AUTO_ARCHIVE_MODE", "per_chain", str),  # per_chain | per_dataset
    "HIVE_ARCHIVE_PATH": ENV("HIVE_ARCHIVE_PATH", "curves.tar.gz", str),
    # staged ingestion chaining (auto-run multiple stages this boot)
    "HIVE_INGEST_CHAIN": ENV("HIVE_INGEST_CHAIN", "1", bool),
    "HIVE_INGEST_CHAIN_MAX": ENV("HIVE_INGEST_CHAIN_MAX", "2", int),  # max stages per boot
    # staged ingestion controls
    "HIVE_INGEST_STAGED": ENV("HIVE_INGEST_STAGED", "1", bool),
    "HIVE_INGEST_STAGE_SIZE": ENV("HIVE_INGEST_STAGE_SIZE", "3", int),
    "HIVE_INGEST_MIN_FREE_GB": ENV("HIVE_INGEST_MIN_FREE_GB", "8", int),
    "HIVE_INGEST_NEXT": ENV("HIVE_INGEST_NEXT", "0", bool),

    # self-edit manifest controls 
    "HIVE_ALLOW_SELF_WRITE_MANIFEST": ENV("HIVE_ALLOW_SELF_WRITE_MANIFEST", "1", bool),
    "HIVE_SELF_WRITE_FILE": ENV("HIVE_SELF_WRITE_FILE", "", str),

    # memory auto-restore controls (admin memory)
    "CURVES_AUTO_RESTORE": ENV("HIVE_CURVES_AUTO_RESTORE", "1", bool),
    "CURVES_ARCHIVE_LOCAL": ENV("HIVE_CURVES_ARCHIVE_LOCAL", "curves.tar.gz", str),
    "CURVES_ARCHIVE_URL": ENV("HIVE_CURVES_ARCHIVE_URL", "", str),
    "CURVES_HF_DATASET": ENV("HIVE_CURVES_HF_DATASET", "", str),
    "CURVES_HF_SUBPATH": ENV("HIVE_CURVES_HF_SUBPATH", "", str),
    "HF_READ_TOKEN": ENV("HF_READ_TOKEN", "", str),

    # memory directory alias
    "HIVE_HOME": ENV("HIVE_HOME", "/home/hive/hive_data" if os.path.exists("/home/hive") else "./hive_data"), # type: ignore
    "CURVE_DIR": os.path.join(ENV("HIVE_HOME", "/home/hive/hive_data" if os.path.exists("/home/hive") else "./hive_data"), "curves"), # type: ignore
    "STATE_DIR": os.path.join(ENV("HIVE_HOME", "/home/hive/hive_data" if os.path.exists("/home/hive") else "./hive_data"), "system"), # type: ignore
    "LAUNCH_UI": ENV("HIVE_LAUNCH_UI","1",bool),
    "LLM_AUTOSIZE": ENV("HIVE_LLM_AUTOSIZE", "1", bool), # type: ignore
    "LLM_MAX_VRAM_GB": ENV("HIVE_LLM_MAX_VRAM_GB","0", int),
    "MODEL_OVERRIDE": ENV("HIVE_MODEL_ID",""),
    "CTX_TOKENS": ENV("HIVE_CTX_TOKENS","2048",int),
    "OWNER_NAME": ENV("HIVE_OWNER_USER","Rose"),
    "OWNER_PASS": ENV("HIVE_OWNER_PASS","Fehr2008"),
    "OWNER_SECOND": ENV("HIVE_OWNER_SECOND","Paulbear01"),
    "AGENT_NAME": ENV("HIVE_AGENT_NAME","Hive"),
    "NO_PROFANITY": ENV("HIVE_NO_PROFANITY","1",bool),
    "ASR_SIZE": ENV("HIVE_ASR_SIZE","small"),
    "TTS_LANG": ENV("HIVE_TTS_LANG","en"),
    "BOOTSTRAP_INGEST": ENV("HIVE_BOOTSTRAP_INGEST","1",bool),
    "FORCE_REINGEST": ENV("HIVE_FORCE_REINGEST","0",bool),
    "INGEST_SOURCES": ENV("HIVE_INGEST_SOURCES",""),
    "ONLINE_ENABLE": ENV("HIVE_ONLINE_ENABLE","1",bool),
    "ONLINE_AUTO": ENV("HIVE_ONLINE_AUTO","0",bool),
    "ONLINE_SOURCES": ENV("HIVE_ONLINE_SOURCES","https://hnrss.org/frontpage,https://rss.nytimes.com/services/xml/rss/nyt/World.xml"),
    "ONLINE_TIMEOUT": ENV("HIVE_ONLINE_TIMEOUT","8",int),
    "ONLINE_MAX_RESULTS": ENV("HIVE_ONLINE_MAX_RESULTS","5",int),
    "ONLINE_TRIGGER": ENV("HIVE_ONLINE_TRIGGER","auto",str),
    # bounded self governance
    "HIVE_USE_HF_INFERENCE": ENV("HIVE_USE_HF_INFERENCE","0",bool),
    "HIVE_HF_ENDPOINT": ENV("HIVE_HF_ENDPOINT","",str),
    "ALLOW_SELF_REBOOT": ENV("HIVE_ALLOW_SELF_REBOOT","1",bool),
    "ALLOW_RUNTIME_HOTPATCH": ENV("HIVE_ALLOW_RUNTIME_HOTPATCH", "1", bool),
    "AUTO_SELF_OPTIMIZE": ENV("HIVE_AUTO_SELF_OPTIMIZE","1",bool),
    "PVPORCUPINE_ACCESS_KEY": ENV("HIVE_PVPORCUPINE_ACCESS_KEY", "", str),
    "HIVE_WAKE_WORDS": ENV("HIVE_WAKE_WORDS", "bumblebee", str), # Default wake word
    "VIDEO_ENABLED": ENV("HIVE_VIDEO_ENABLED", "0", bool), # Add this line
    # internal optimization with sandbox + A/B (Owner policy)
    "OPT_ENABLE": ENV("HIVE_OPT_ENABLE","1",bool),
    "OPT_AUTO_APPLY": ENV("HIVE_OPT_AUTO_APPLY","0",bool),  # OWNER MAY SET TO 1
    "OPT_PKG_ALLOWLIST": ENV("HIVE_OPT_PKG_ALLOWLIST","transformers,accelerate,datasets,sentence-transformers,faiss-cpu,duckduckgo_search,feedparser,requests,gradio").split(","),
    "OPT_MODEL_ALLOWLIST": ENV("HIVE_OPT_MODEL_ALLOWLIST","meta-llama/Meta-Llama-3.1-8B-Instruct,meta-llama/Meta-Llama-3.1-70B-Instruct,TinyLlama/TinyLlama-1.1B-Chat-v1.0").split(","),
    "OPT_THRESH_LATENCY_MS": ENV("HIVE_OPT_THRESH_LATENCY_MS","0",int),
    "OPT_THRESH_TOKS_PER_S": ENV("HIVE_OPT_THRESH_TOKS_PER_S","0",float),
    "OPT_THRESH_QUALITY": ENV("HIVE_OPT_THRESH_QUALITY","0.02",float),
    "OPT_SANDBOX_TIMEOUT": ENV("HIVE_OPT_SANDBOX_TIMEOUT","180",int),
}
CFG["VOICE_ASR_MODEL"] = CFG["ASR_SIZE"] # Alias for backward compatibility

HIVE_INSTANCE = None

CFG['VAD_ENERGY_THRESHOLD'] = 300
CFG['VAD_SILENCE_DURATION'] = 1.0
CFG['VAD_MIN_SPEECH_DURATION'] = 0.2
CFG['VOICE_VAD_AGGRESSIVENESS'] = 2 # Default VAD aggressiveness

# Create all necessary directories based on the new specification 
HIVE_HOME = CFG["HIVE_HOME"] # type: ignore
DIRS_TO_CREATE = [
    os.path.join(HIVE_HOME, "curves"),
    os.path.join(HIVE_HOME, "knowledge", "chunks"),
    os.path.join(HIVE_HOME, "knowledge", "embeddings"),
    os.path.join(HIVE_HOME, "users", "conversations"),
    os.path.join(HIVE_HOME, "users", "sessions"),
    os.path.join(HIVE_HOME, "system", "logs"),
    os.path.join(HIVE_HOME, "system", "backups"),
    os.path.join(HIVE_HOME, "voice", "asr_models"),
    os.path.join(HIVE_HOME, "voice", "tts_models"),
    os.path.join(HIVE_HOME, "voice", "voiceprints"),
    os.path.join(HIVE_HOME, "voice", "samples"),
    os.path.join(HIVE_HOME, "admin", "logs"),
    os.path.join(HIVE_HOME, "packages"),
]
for d in DIRS_TO_CREATE: os.makedirs(d, exist_ok=True)

OVERLAY_DIR = os.path.join(HIVE_HOME, "system", "overlay")
OPT_DIR = os.path.join(HIVE_HOME, "system", "opt")
OPT_PROPOSALS = os.path.join(OPT_DIR, "proposals.jsonl")
OPT_RESULTS   = os.path.join(OPT_DIR, "results.jsonl")
for p in (OVERLAY_DIR, OPT_DIR):
    os.makedirs(p, exist_ok=True)

# ----------------- sensing / model pick -----------------
class EnvDetector:
    """Implements the Environment Detector and Capability Profiler from Part 1, Section 4."""
    def _has_gpu_env(self) -> bool:
        accel = os.getenv("SPACE_ACCELERATOR", "").lower()
        if accel in ("t4", "a10", "a100", "l4", "l40", "h100"): return True
        try:
            return torch is not None and torch.cuda.is_available()
        except Exception:
            return False

    def _detect_display(self) -> bool:
        if _os_name() == 'linux':
            return bool(os.environ.get('DISPLAY')) or os.path.exists('/dev/fb0')
        return False # Simplified for other OSes

    def _detect_camera(self) -> bool:
        if _os_name() == 'linux':
            return any(os.path.exists(f'/dev/video{i}') for i in range(4))
        return False

    def _detect_audio_input(self) -> bool:
        # This is a heuristic; a more robust check would use sounddevice or similar
        return True

    def probe(self) -> Dict[str, any]:
        total_ram_gb = psutil.virtual_memory().total / (1024**3)
        is_pi = 'raspberrypi' in platform.machine().lower()
        profile = {
            "device_type": "raspberry_pi" if is_pi else "generic_linux",
            "arch": platform.machine(),
            "total_ram_gb": round(total_ram_gb, 1),
            "free_ram_gb": round(psutil.virtual_memory().available / (1024**3), 1),
            "has_gpu": self._has_gpu_env(),
            "has_display": self._detect_display(),
            "has_camera": self._detect_camera(),
            "has_microphone": self._detect_audio_input(),
            "network_up": NET.online_quick(),
            "is_low_memory": total_ram_gb < 6,
            "max_docs": 70000 if total_ram_gb > 16 else (50000 if total_ram_gb > 8 else 12000),
            "batch": 512 if total_ram_gb > 16 else (256 if total_ram_gb > 8 else 64)
        }
        return profile

def probe_caps():
    return EnvDetector().probe()

CANDIDATES=[("TinyLlama/TinyLlama-1.1B-Chat-v1.0",0),("meta-llama/Meta-Llama-3.1-8B-Instruct",12),("meta-llama/Meta-Llama-3.1-70B-Instruct",100)]
def pick_model(caps: Dict[str, any]) -> Tuple[str, dict]: # type: ignore
    """Always selects TinyLlama for simplicity in this version."""
    model_id = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"
    device = "cuda" if _has_gpu_env() else "cpu"
    return model_id, {"device": device}

# ----------------- embeddings / curves -----------------
_EMB_ID=os.getenv("HIVE_EMB_ID","sentence-transformers/all-MiniLM-L6-v2")
class GEC:
    def __init__(self):
        device = "cuda" if EnvDetector()._has_gpu_env() else "cpu"
        self.model=SentenceTransformer(_EMB_ID).to(device)
    def encode(self, texts: List[str]): return self.model.encode(texts, normalize_embeddings=True)

class CurveStore:
    def __init__(self, d):
        self.dir=d; os.makedirs(d, exist_ok=True)
        self.idx_path=os.path.join(d,"faiss.index")
        self.meta_path=os.path.join(d,"meta.jsonl")
        self.dim=384; self.gec=GEC()
        self.index=faiss.read_index(self.idx_path) if os.path.exists(self.idx_path) else faiss.IndexFlatIP(self.dim)
    def add_texts(self, docs:List[str], metas:List[Dict]):
        # This is the old, direct-to-FAISS method. It will be deprecated by the new KnowledgeStore.
        # For now, we keep it for compatibility with existing code paths but new ingestion should use KnowledgeStore.
        # The new KnowledgeStore will handle chunking, manifest updates, and background embedding.
        # This method will be refactored to be a part of the background embedding worker.
        if not docs: return
        vecs=np.asarray(self.gec.encode(docs), dtype="float32")
        self.index.add(vecs)
        with open(self.meta_path,"a",encoding="utf-8") as f:
            for m in metas: f.write(json.dumps(m, ensure_ascii=False)+"\n")
        faiss.write_index(self.index, self.idx_path)
    def search(self, query:str, k:int=6)->List[Dict]:
        if self.index.ntotal==0: return []
        qv=np.asarray(self.gec.encode([query]), dtype="float32")
        D,I=self.index.search(qv,k)
        lines=open(self.meta_path,"r",encoding="utf-8").read().splitlines() if os.path.exists(self.meta_path) else []
        out=[]
        for i in I[0]:
            if 0<=i<len(lines):
                try: out.append(json.loads(lines[i])) # type: ignore
                except json.JSONDecodeError: pass # type: ignore
        return out
    def search_with_scores(self, query:str, k:int=6):
        if self.index.ntotal == 0: return [], []
        qv=np.asarray(self.gec.encode([query]), dtype="float32")
        D,I=self.index.search(qv,k) # type: ignore
        lines=open(self.meta_path,"r",encoding="utf-8").read().splitlines() if os.path.exists(self.meta_path) else []
        metas, scores = [], [] # type: ignore
        query_len = len(query.split())

        for idx, sc in zip(I[0], D[0]):
            if 0<=idx<len(lines):
                try:
                    meta = json.loads(lines[idx])
                    # Penalize long snippets for short queries to avoid irrelevant context.
                    text_len = len(meta.get("text", "").split())
                    penalty = 0.0
                    if query_len < 4 and text_len > 100:
                        penalty = 0.15 * (min(text_len, 400) / 400) # Penalize up to 0.15
                    
                    metas.append(meta)
                    scores.append(float(max(0.0, min(1.0, (sc if sc is not None else 0.0) - penalty)))) # type: ignore
                except: pass
        return metas, scores

OFFLINE_MARK = os.path.join(CFG["CURVE_DIR"], ".offline_ready")
def _curves_ready(curve_dir:str)->bool:
    idx=os.path.join(curve_dir,"faiss.index")
    if os.path.exists(OFFLINE_MARK):
        try: return json.load(open(OFFLINE_MARK)).get("ok",True)
        except Exception: return True
    if os.path.exists(idx):
        try: return faiss.read_index(idx).ntotal>0
        except Exception: return False
    return False
def _mark_offline_ready():
    try: json.dump({"ok":True,"ts":time.time()}, open(OFFLINE_MARK,"w",encoding="utf-8"))
    except Exception: pass

# ----------- HF Datasets bootstrap -----------
DEFAULT_SOURCES=["jhu-clsp/jflue","bea2019st/wi_locness","fce-m2109/mascorpus","rajpurkar/squad_v2",
                 "OpenRL/daily_dialog","tetti/spelling-dataset-extended","Helsinki-NLP/opus-100","facebook/flores",
                 "HuggingFaceH4/no_robots","bigscience/xP3","allenai/sciq","allenai/c4",
                 "mozilla-foundation/common_voice_17_0","bene-ges/en_cmudict","openslr/librispeech_asr","conceptnet5/conceptnet5","grammarly/coedit"]

def _atomic_write_json(path, data):
    tmp = str(path) + f".tmp_{int(time.time())}"
    with open(tmp, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    os.replace(tmp, path)

def _load_json(path, default):
    if os.path.exists(path):
        try:
            with open(path, "r", encoding="utf-8") as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError):
            return default
    return default

def _save_json(path, data):
    # This function is not defined in the provided code. Assuming it should be _atomic_write_json
    _atomic_write_json(path, data)

class KnowledgeStore:
    def __init__(self, storage_path: str):
        self.base = _Path(storage_path)
        self.knowledge_dir = self.base / "knowledge"
        self.chunks_dir = self.knowledge_dir / "chunks"
        self.curves_dir = self.base / "curves"
        for d in [self.knowledge_dir, self.chunks_dir, self.curves_dir]:
            d.mkdir(parents=True, exist_ok=True)

        self.manifest_path = self.knowledge_dir / "knowledge_manifest.json"
        self.embedding_queue_path = self.knowledge_dir / "embedding_queue.jsonl"
        self._lock = threading.RLock()
        self._load_manifest()

    def _load_manifest(self):
        with self._lock:
            if self.manifest_path.exists():
                try:
                    with open(self.manifest_path, 'r', encoding='utf-8') as f:
                        self.manifest = json.load(f)
                except json.JSONDecodeError:
                    self.manifest = self._default_manifest()
            else:
                self.manifest = self._default_manifest()
                self._save_manifest()

    def _default_manifest(self):
        return {
            "total_chunks": 0, "total_texts": 0, "chunks_by_tag": {},
            "chunks_by_scope": {}, "chunk_index": {}, "last_vector_build": 0,
            "vector_count": 0
        }

    def _save_manifest(self):
        with self._lock:
            _atomic_write_json(self.manifest_path, self.manifest)

    def _normalize_text(self, text: str) -> str:
        return unicodedata.normalize("NFC", text).strip()

    def _chunk_text(self, text: str, target_size: int = 1000) -> List[str]:
        # Simple sentence-based chunking for now.
        sentences = re.split(r'(?<=[.!?])\s+', text)
        chunks, current_chunk = [], ""
        for sentence in sentences:
            if len(current_chunk) + len(sentence) + 1 > target_size:
                if current_chunk: chunks.append(current_chunk)
                current_chunk = sentence
            else:
                current_chunk += (" " + sentence) if current_chunk else sentence
        if current_chunk: chunks.append(current_chunk)
        return chunks

    def ingest_text(self, text: str, tag: str="ingest", scope: str="general", metadata: Optional[Dict]=None) -> Optional[str]:
        with self._lock:
            normalized = self._normalize_text(text)
            if not normalized: return None
            
            texts = self._chunk_text(normalized)
            if not texts: return None

            chunk_id = f"chunk_{int(time.time())}_{hashlib.sha1(texts[0].encode('utf-8')).hexdigest()[:8]}"
            chunk_data = {
                "chunk_id": chunk_id, "timestamp": time.time(), "tag": tag, "scope": scope,
                "text_count": len(texts), "texts": texts, "metadata": metadata or {},
                "quality_score": 0.7, "importance_score": 0.5, # Defaults
                "embeddings_generated": False
            }
            chunk_file = self.chunks_dir / f"{chunk_id}.json"
            _atomic_write_json(chunk_file, chunk_data)

            # Update manifest
            self.manifest["total_chunks"] += 1
            self.manifest["total_texts"] += len(texts)
            self.manifest.setdefault("chunks_by_tag", {}).setdefault(tag, []).append(chunk_id)
            self.manifest.setdefault("chunks_by_scope", {}).setdefault(scope, []).append(chunk_id)
            self.manifest.setdefault("chunk_index", {})[chunk_id] = {
                "timestamp": chunk_data["timestamp"], "tag": tag, "scope": scope,
                "text_count": len(texts), "quality_score": chunk_data["quality_score"]
            }
            self._save_manifest()

            # Enqueue for embedding
            with open(self.embedding_queue_path, "a", encoding="utf-8") as f:
                f.write(json.dumps({"chunk_id": chunk_id, "status": "queued"}) + "\n")
            
            return chunk_id

# ----------- voice: ASR/TTS/phonics -----------
G2P = G2p()
class ASRService:
    """Handles ASR, including transcription and language detection."""
    def __init__(self):
        # This will be initialized in the VoiceServicesModule
        self.model = get_asr()

    def transcribe(self, audio_path: str, uid: Optional[str], forced_lang: Optional[str] = None) -> dict:
        prior = _load_json(ADAPT_DB, {}).get(uid or "guest", {}).get("lang_prior")
        language = forced_lang or prior or None
        # Assuming get_asr() returns a valid model object
        segs, info = self.model.transcribe(audio_path, language=language, beam_size=5, vad_filter=True)
        text = " ".join([s.text for s in segs]).strip()

        detected_lang = info.language
        if not forced_lang and text:
            prof = _load_json(ADAPT_DB, {})
            p = prof.get(uid or "guest", {})
            p["lang_prior"] = detected_lang
            prof[uid or "guest"] = p
            _save_json(ADAPT_DB, prof)

        return {"text": text, "language": detected_lang, "confidence": info.language_probability, "segments": [{"start": s.start, "end": s.end, "text": s.text} for s in segs]}

ASR_MODELS={"tiny":"tiny","base":"base","small":"small","medium":"medium","large":"large-v3"}
def _asr_model_name(): return ASR_MODELS.get(CFG["VOICE_ASR_MODEL"],"small")
_ASR=None
def get_asr():
    global _ASR
    if _ASR is not None: return _ASR
    size=_asr_model_name(); device="cuda" if (_has_gpu_env()) else "cpu"
    compute_type="float16" if device=="cuda" else "int8"
    _ASR=WhisperModel(size, device=device, compute_type=compute_type); return _ASR

PIPER_MODELS={
    "en": ("https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/amy/low/en_US-amy-low.onnx",
           "https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/amy/low/en_US-amy-low.onnx.json"),
    "es": ("https://huggingface.co/rhasspy/piper-voices/resolve/main/es/es_ES/davefx/medium/es_ES-davefx-medium.onnx",
           "https://huggingface.co/rhasspy/piper-voices/resolve/main/es/es_ES/davefx/medium/es_ES-davefx-medium.onnx.json"),
    "fr": ("https://huggingface.co/rhasspy/piper-voices/resolve/main/fr/fr_FR/gilles/medium/fr_FR-gilles-medium.onnx",
           "https://huggingface.co/rhasspy/piper-voices/resolve/main/fr/fr_FR/gilles/medium/fr_FR-gilles-medium.onnx.json"),
    "de": ("https://huggingface.co/rhasspy/piper-voices/resolve/main/de/de_DE/thorsten-deepbinner/low/de_DE-thorsten-deepbinner-low.onnx",
           "https://huggingface.co/rhasspy/piper-voices/resolve/main/de/de_DE/thorsten-deepbinner/low/de_DE-thorsten-deepbinner-low.onnx.json"),
    "zh": ("https://huggingface.co/rhasspy/piper-voices/resolve/main/zh/zh_CN/huayan/low/zh_CN-huayan-low.onnx",
           "https://huggingface.co/rhasspy/piper-voices/resolve/main/zh/zh_CN/huayan/low/zh_CN-huayan-low.onnx.json"),
}
def _download(url,dst, timeout=30): # type: ignore
    if os.path.exists(dst): return dst
    os.makedirs(os.path.dirname(dst),exist_ok=True); urllib.request.urlretrieve(url,dst); return dst # TODO: add timeout
_TTS_CACHE={}
def get_tts(lang: str = "en") -> PiperVoice: # type: ignore
    lang=lang if lang in PIPER_MODELS else "en"
    if lang in _TTS_CACHE: return _TTS_CACHE[lang] 
    mu,cu=PIPER_MODELS[lang]; m=_download(mu,f"./models/piper/{os.path.basename(mu)}"); c=_download(cu,f"./models/piper/{os.path.basename(cu)}")
    v=PiperVoice.load(m,c); _TTS_CACHE[lang]=v; return v

def _embed_mfcc(path)->np.ndarray:
    y, sr = librosa.load(path, sr=16000)
    mf=librosa.feature.mfcc(y=y, sr=sr, n_mfcc=20)
    return mf.mean(axis=1)
def enroll_voice(uid:str, path:str) -> bool:
    db=_load_json(VOICES_DB, {}); db[uid]=_embed_mfcc(path).astype(float).tolist(); _save_json(VOICES_DB, db); return True
def identify_voice(path:str, threshold:float=0.70) -> Optional[str]:
    db=_load_json(VOICES_DB, {}); 
    if not db: return None
    emb=_embed_mfcc(path).reshape(1,-1)
    keys=list(db.keys()); mats=np.array([db[k] for k in keys])
    sims=cosine_similarity(emb, mats)[0]; i=int(np.argmax(sims)); return keys[i] if sims[i]>=threshold else None

_BASIC={'a':'a as in apple /æ/','e':'e as in elephant /ɛ/','i':'i as in igloo /ɪ/','o':'o as in octopus /ɒ/','u':'u as in umbrella /ʌ/',
        'c':'c as in cat /k/ (before e/i/y often /s/)','g':'g as in goat /g/ (before e/i/y often soft /dʒ/)','y':'y as in yellow /j/ or happy /i/'}
def phonics(word:str)->str:
    toks=G2P(word); phones=[t for t in toks if re.match(r"[A-Z]+[0-2]?$", t)]
    hints=[]; 
    for ch in word.lower():
        if ch in _BASIC and _BASIC[ch] not in hints: hints.append(_BASIC[ch])
    return f"Phonemes: {' '.join(phones)} | Hints: {('; '.join(hints)) if hints else '🐝'}"

def lid_chunk(text:str, min_len:int=12)->List[Tuple[str,str]]:
    parts=re.split(r"([.!?;\u2026\u2028\u2029])+\s{2,}|", text)
    chunks=[]; buf=""
    for p in parts:
        if not p: continue
        buf+=p
        if len(buf)>=min_len or re.match(r"[.!?;\u2026\u2028\u2029]", p):
            lang,_=langid.classify(buf.strip()); chunks.append((buf.strip(), lang)); buf=""
    if buf.strip():
        lang,_=langid.classify(buf.strip()); chunks.append((buf.strip(), lang))
    return chunks

def asr_transcribe(path:str, uid: Optional[str], forced_lang: Optional[str]=None)->str:
    # This function seems to duplicate ASRService.transcribe logic.
    # It's better to use the service.
    model=get_asr()
    prior=_load_json(ADAPT_DB,{}).get(uid or "guest",{}).get("lang_prior")
    language=forced_lang or prior or None
    segs, info = model.transcribe(path, language=language, beam_size=5, vad_filter=True)
    text=" ".join([s.text for s in segs]) if segs else ""
    if not forced_lang and text.strip(): # type: ignore
        lid,_=langid.classify(text); prof=_load_json(ADAPT_DB,{}); p=prof.get(uid or "guest",{}); p["lang_prior"]=lid; prof[uid or "guest"]=p; _save_json(ADAPT_DB,prof)
    return text

def synthesize_multilang(text:str, fallback="en")->str:
    # This function is now simplified as the TTSService handles caching and logic.
    v = get_tts(fallback)
    aud, _ = v.synthesize(text)
    sr = v.sample_rate
    mix = aud
    outp=os.path.join(tempfile.gettempdir(), f"hive_tts_{int(time.time())}.wav")
    sf.write(outp, mix if mix is not None else np.zeros(1), sr or 22050, subtype="PCM_16"); return outp

# ----------- compiler / engine -----------

class EngineCurve:
    def __init__(self):
        self.stats={"runs":0,"ok":0,"latency_ms":[]}
        self.router_rules=[]
    def choose_route(self, msg:str)->str:
        # This is a simplified version. The full logic is now in IntentRouter.
        return "tutor"
    def run(self, message:str, snippets:List[Dict])->Dict: return {"ok":True,"route":"tutor"}
# ----------- wifi auto-connect (non-blocking) -----------
NET_STATE_DB=os.path.join(CFG["STATE_DIR"],"wifi_known.json")

def _os_name(): return platform.system().lower()
def _fast_probe(host="8.8.8.8", port=53, timeout=1.5) -> bool:
    try:
        socket.setdefaulttimeout(timeout)
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM); s.connect((host, port)); s.close()
        return True
    except Exception:
        return False
def _http_probe(url="https://huggingface.co", timeout=2.5)->float:
    try:
        t0=time.time(); r=requests.head(url, timeout=timeout)
        if r.status_code<500: return (time.time()-t0)*1000.0
    except Exception: pass
    return -1.0
def _load_known()->List[dict]:
    data=_load_json(NET_STATE_DB, []); out=[]
    for d in data:
        if isinstance(d,dict) and "ssid" in d:
            out.append({"ssid":d["ssid"],"priority":int(d.get("priority",0))})
    out.sort(key=lambda x: x.get("priority",0), reverse=True); return out
def _get_saved_password(ssid:str)->Optional[str]:
    if keyring:
        try: return keyring.get_password("hive_wifi", ssid) or "" # type: ignore
        except Exception: return None
    return None
def _connect_linux(ssid, password, timeout=12)->Tuple[bool,str]:
    try:
        cmd=["nmcli","device","wifi","connect",ssid]+(["password",password] if password else [])
        p=subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        return (p.returncode==0), (p.stdout or p.stderr or "").strip()
    except Exception as e: return False, f"nmcli error: {e}"
def _connect_windows(ssid, password)->Tuple[bool,str]:
    try:
        p=subprocess.run(["netsh","wlan","connect","name="+ssid,"ssid="+ssid], capture_output=True, text=True)
        if p.returncode==0 and "success" in (p.stdout+p.stderr).lower(): return True,"Connected."
        if not password: return False,"No saved password."
        xml=f'''<?xml version="1.0"?>
<WLANProfile xmlns="http://www.microsoft.com/networking/WLAN/profile/v1">
  <name>{ssid}</name><SSIDConfig><SSID><name>{ssid}</name></SSIDConfig>
  <connectionType>ESS</connectionType><connectionMode>auto</connectionMode>
  <MSM><security><authEncryption><authentication>WPA2PSK</authentication>
  <encryption>AES</encryption><useOneX>false</useOneX></authEncryption>
  <sharedKey><keyType>passPhrase</keyType><protected>false</protected>
  <keyMaterial>{password}</keyMaterial></sharedKey></security></MSM></WLANProfile>'''
        tmp=os.path.join(os.getenv("TEMP","/tmp"), f"wifi_{int(time.time())}.xml"); open(tmp,"w",encoding="utf-8").write(xml)
        a=subprocess.run(["netsh","wlan","add","profile","filename="+tmp,"user=all"], capture_output=True, text=True)
        if a.returncode!=0: return False, a.stderr or a.stdout or "add profile failed"
        c=subprocess.run(["netsh","wlan","connect","name="+ssid,"ssid="+ssid], capture_output=True, text=True)
        return (c.returncode==0), (c.stderr or c.stdout or "").strip()
    except Exception as e: return False, f"netsh error: {e}"
def _connect_macos(ssid, password)->Tuple[bool,str]:
    try:
        out=subprocess.check_output(["networksetup","-listallhardwaresports"], stderr=subprocess.DEVNULL).decode("utf-8","ignore")
        dev=None
        for block in out.split("\n\n"):
            if "Wi-Fi" in block or "AirPort" in block:
                for l in block.splitlines():
                    if l.strip().startswith("Device:"): dev=l.split(":",1)[1].strip(); break
                if dev: break
        if not dev: return False,"Wi-Fi device not found"
        cmd=["networksetup","-setairportnetwork",dev, ssid]+([password] if password else [])
        p=subprocess.run(cmd, capture_output=True, text=True)
        return (p.returncode==0), (p.stderr or p.stdout or "").strip()
    except Exception as e: return False, f"networksetup error: {e}"
def _connect_os(ssid,password,timeout=12)->Tuple[bool,str]:
    osn=_os_name()
    if osn=="linux": return _connect_linux(ssid,password,timeout)
    if osn=="windows": return _connect_windows(ssid,password)
    if osn=="darwin": return _connect_macos(ssid,password)
    return False, f"Unsupported OS: {osn}"

class AutoConnector:
    def __init__(self):
        self.last_attempt=0.0; self.cooldown_s=30.0; self.per_ssid_timeout=10.0; self.total_budget_s=18.0; self.thread=None; self._lock=threading.Lock()
    def online_quick(self)->bool: return _fast_probe(timeout=1.2)
    def quality_ms(self)->float: return _http_probe(timeout=2.0)
    def _run_once(self):
        if self.online_quick(): return
        known=_load_known(); 
        if not known: return
        t_start=time.time()
        for item in known:
            if time.time()-t_start>self.total_budget_s: return
            ssid=item["ssid"]; pw=_get_saved_password(ssid)
            ok,_msg=_connect_os(ssid,pw,timeout=int(self.per_ssid_timeout))
            if ok and self.online_quick(): return
    def kick_async(self):
        with self._lock:
            now=time.time()
            if now - self.last_attempt < self.cooldown_s: return
            self.last_attempt=now
            if self.thread and self.thread.is_alive(): return
            self.thread = threading.Thread(target=self._run_once, daemon=True); self.thread.start()

NET = AutoConnector()

def _has_gpu_env() -> bool:
    """Global helper to check for GPU environment."""
    return EnvDetector()._has_gpu_env()


# ----------- coverage heuristic -----------
def coverage_score_from_snippets(snippets: list, scores: list) -> float:
    if not snippets or not scores: return 0.0
    s = sorted(scores, reverse=True)[:3]
    base = sum(s) / len(s) if s else 0.0 # type: ignore
    bonus = min(0.15, 0.03 * len(snippets))
    return float(max(0.0, min(1.0, base + bonus)))

# ----------- RBAC / users / lockouts (Restored) -----------
USERS_DB=os.path.join(CFG["STATE_DIR"],"users.json")
LOCKS_DB=os.path.join(CFG["STATE_DIR"],"lockouts.json")
VOICES_DB=os.path.join(CFG["STATE_DIR"],"voices.json")
ADAPT_DB=os.path.join(CFG["STATE_DIR"],"speech_adapt.json")

def _init_users():
    d={"owner":{"id":"owner:1","name":CFG["OWNER_NAME"],"role":"owner","pass":CFG["OWNER_PASS"],"second":CFG["OWNER_SECOND"],"prefs":{"activation_names":[CFG["AGENT_NAME"]],"language":"en"}},
       "admins_super":[],"admins_general":[],"users":[]}
    _save_json(USERS_DB,d); return d
def _load_users():
    d=_load_json(USERS_DB, None); return d if d else _init_users()
def _find_user(d, name_or_id):
    pools=[("owner",[d.get("owner")]),("admin_super",d.get("admins_super", [])),("admin_general",d.get("admins_general", [])),("user",d.get("users", []))]
    for role,pool in pools:
        for u in pool or []:
            if u and (u.get("id")==name_or_id or u.get("name")==name_or_id): return u, role
    return None, None

PERMS={
    "owner":{"can_add":["admin_super","admin_general","user"],"can_remove":["admin_super","admin_general","user"],
             "can_edit_role_of":["admin_super","admin_general","user"],"can_edit_profile_of":["owner","admin_super","admin_general","user"],
             "can_view_scopes":"all","maintenance":"full","code_edit":"approve_and_edit"},
    "admin_super":{"can_add":["admin_general","user"],"can_remove":["admin_general","user"],
             "can_edit_role_of":["admin_general","user"],"can_edit_profile_of":["admin_general","user"],
             "can_view_scopes":"self_only","maintenance":"advanced","code_edit":"suggest_only"},
    "admin_general":{"can_add":["user"],"can_remove":["user"],"can_edit_role_of":["user"],"can_edit_profile_of":["user"],
             "can_view_scopes":"self_only","maintenance":"basic","code_edit":"suggest_only"},
    "user":{"can_add":[],"can_remove":[],"can_edit_role_of":[],"can_edit_profile_of":["user"],
             "can_view_scopes":"self_only","maintenance":"none","code_edit":"none"},
    "guest":{"can_add":[],"can_remove":[],"can_edit_role_of":[],"can_edit_profile_of":[],
             "can_view_scopes":"self_only","maintenance":"none","code_edit":"none"},
}

def attempt_login(name_or_id:str, password:str="", second:Optional[str]=None):
    d=_load_users(); locks=_load_json(LOCKS_DB,{ })
    def lock_fail(lid, msg):
        st=locks.get(lid, {"fails":0,"until":0}); st["fails"]=st.get("fails",0)+1; dur=180 if st["fails"]>=3 else 0; st["until"]=time.time()+dur if dur else 0
        locks[lid]=st; _save_json(LOCKS_DB,locks); return False, msg
    u,_=_find_user(d, name_or_id)
    if not u: return False, "Profile not found."
    role=u.get("role","user"); lid=str(u.get("id", u.get("name"))); now=time.time(); st=locks.get(lid, {"fails":0,"until":0})
    if now < st.get("until",0): return False, f"Locked; try again in ~{int(st['until']-now)}s."
    if role in ("admin_general","admin_super","owner") and (password!=u.get("pass") or (role=="owner" and u.get("second") and second!=u.get("second"))): return lock_fail(lid, "Credentials incorrect.")
    locks[lid]={"fails":0,"until":0}; _save_json(LOCKS_DB,locks); return True, f"Welcome, {u.get('name')} ({role})."

# ----------- overlay / hotpatch -----------
RUNTIME_OVERRIDES = os.path.join(HIVE_HOME, "system", "runtime_overrides.json")
ALLOWED_PATCH_KEYS={"prompt_head","retrieval_k","token_budget","temperature","router_rules","web_threshold"}
def _load_overrides():
    if os.path.exists(RUNTIME_OVERRIDES):
        try: return json.load(open(RUNTIME_OVERRIDES,"r",encoding="utf-8"))
        except Exception: return {}
    return {}
def _save_overrides(ovr:dict):
    _atomic_write_json(RUNTIME_OVERRIDES, ovr)

class RuntimeOverlay:
    def __init__(self): self.ovr=_load_overrides()
    def apply_to(self, hive: "Hive"):
        o=self.ovr or {}
        if isinstance(o.get("prompt_head"),str): hive.compiler.override_head=o["prompt_head"]
        if isinstance(o.get("token_budget"),int): hive.compiler.override_budget=max(256, min(8192, o["token_budget"]))
        hive.retrieval_k=int(o.get("retrieval_k",6)); hive.retrieval_k=max(3,min(24,hive.retrieval_k))
        hive.decoding_temperature=float(o.get("temperature",0.7)); hive.decoding_temperature=max(0.0,min(1.5,hive.decoding_temperature))
        rr=o.get("router_rules") or []
        if isinstance(rr,list):
            try: hive.engine.router_rules=[re.compile(pat,re.I) for pat in rr if isinstance(pat,str) and pat]
            except re.error: hive.engine.router_rules=[]
        t=o.get("web_threshold",None); hive.web_threshold=float(t) if isinstance(t,(int,float)) else 0.40
    def patch(self, patch:dict, actor_role:str="hive")->Tuple[bool,str]:
        if not CFG["ALLOW_RUNTIME_HOTPATCH"]: return False,"Runtime hotpatch disabled."
        if actor_role not in ("hive","admin_general","admin_super","owner"): return False,"Unauthorized actor."
        for k in list(patch.keys()): 
            if k not in ALLOWED_PATCH_KEYS: patch.pop(k,None)
        if not patch: return False,"No allowed keys."
        self.ovr.update(patch); _save_overrides(self.ovr); return True,"Patched."

# ----------- safe reboot -----------
def _persist_before_reboot():
    try: _atomic_write_json(os.path.join(HIVE_HOME, "system", "last_reboot.json"), {"ts":time.time(),"note":"self-reboot"})
    except Exception: pass
def safe_reboot(reason:str="optimization"):
    if not CFG["ALLOW_SELF_REBOOT"]: return False,"Self-reboot disabled."
    _persist_before_reboot()
    try:
        os.execv(sys.executable, [sys.executable, os.path.abspath(__file__)] + sys.argv[1:])
    except Exception:
        os._exit(3)
    return True, f"Rebooting: {reason}"

# ----------- self optimizer (bounded) -----------
class SelfOptimizer(threading.Thread): # type: ignore
    def __init__(self, hive: "Hive"):
        super().__init__(daemon=True); self.hive=hive; self.stop=False; self.tick=45.0
        self.last_pkg_check = 0
        self.last_code_review = 0
        self.code_review_interval = 3600 * 24  # Check for self-improvement once a day
        self.pkg_check_interval = 3600 * 6 # Check for package updates every 6 hours

    def _check_for_package_updates(self):
        """Checks for updates to packages in the allowlist and proposes changes."""
        if time.time() - self.last_pkg_check < self.pkg_check_interval:
            return
        self.last_pkg_check = time.time()
        print("[SelfOptimizer] Checking for package updates...")
        try:
            # Use pip to check for outdated packages
            outdated_raw = subprocess.check_output([sys.executable, "-m", "pip", "list", "--outdated"], text=True)
            for line in outdated_raw.splitlines()[2:]: # Skip header
                parts = line.split()
                if len(parts) < 3: continue
                pkg_name, current_ver, latest_ver = parts[0], parts[1], parts[2]
                # If the outdated package is in our allowlist, propose an update
                if pkg_name in CFG["OPT_PKG_ALLOWLIST"]:
                    print(f"[SelfOptimizer] Found update for {pkg_name}: {current_ver} -> {latest_ver}")
                    proposal = ChangeProposal(
                        kind="package",
                        name=pkg_name,
                        version=latest_ver,
                        reason=f"Autonomous proposal to update from {current_ver} to {latest_ver}",
                        proposer="hive_optimizer"
                    )
                    proposal_id = self.hive.changes.propose(proposal)
                    # Automatically test the new proposal
                    test_result = self.hive.changes.test_and_compare(proposal_id, proposal)
                    print(f"[SelfOptimizer] Test result for {pkg_name} update: {test_result.get('passed')}, Delta: {test_result.get('delta')}")
        except Exception as e:
            print(f"[SelfOptimizer] Error checking for package updates: {e}")

    def _propose_self_improvement(self):
        """Asks the LLM to review a part of its own code and proposes a change if valid."""
        if time.time() - self.last_code_review < self.code_review_interval:
            return
        self.last_code_review = time.time()
        print("[SelfOptimizer] Performing autonomous code review...")

        try:
            # Read its own source code
            with open(__file__, 'r', encoding='utf-8') as f:
                own_code = f.read()

            # Select a function to review (e.g., coverage_score_from_snippets)
            target_func_name = "coverage_score_from_snippets"
            match = re.search(rf"def {target_func_name}\(.*?^$", own_code, re.S | re.M)
            if not match:
                print(f"[SelfOptimizer] Could not find function {target_func_name} to review.")
                return
            
            func_code = match.group(0)
            prompt = f"""
Review the following Python function for correctness, efficiency, and adherence to best practices.
If you find an improvement, provide ONLY the complete, new, improved function code. Do not add any explanation.
If no improvement is needed, return the original code exactly as it is.

Original function:
```python
{func_code}
```
"""
            # Use the Hive's own chat method to get the LLM's suggestion
            suggested_code = self.hive.chat(prompt, "owner", "hive_optimizer")

            # If the suggestion is different and seems valid, propose it as a code change
            if suggested_code.strip() != func_code.strip() and "def" in suggested_code:
                new_source = own_code.replace(func_code, suggested_code)
                proposal = ChangeProposal(kind="code", name=__file__, patch_text=new_source, reason=f"Autonomous self-improvement of {target_func_name}", proposer="hive_optimizer")
                proposal_id = self.hive.changes.propose(proposal)
                print(f"[SelfOptimizer] Proposing self-improvement change {proposal_id}.")
                test_result = self.hive.changes.test_and_compare(proposal_id, proposal)
                print(f"[SelfOptimizer] Test result for self-improvement: {test_result.get('passed')}, Delta: {test_result.get('delta')}")
        except Exception as e:
            print(f"[SelfOptimizer] Error during self-improvement proposal: {e}")

    def run(self):
        while not self.stop:
            time.sleep(self.tick)
            if not CFG["AUTO_SELF_OPTIMIZE"]: continue

            # --- Autonomous Proposal Generation ---
            self._check_for_package_updates()
            self._propose_self_improvement()

            # --- Real-time Overlay Adjustments ---
            vm=psutil.virtual_memory(); ovr={}
            if vm.percent>88: # type: ignore
                ovr["token_budget"]=max(512,int(0.75*(self.hive.compiler.override_budget or CFG["CTX_TOKENS"]))) # type: ignore
                ovr["temperature"]=max(0.2,self.hive.decoding_temperature-0.1)

            lat=(sum(self.hive.engine.stats["latency_ms"][-10:])/max(1,len(self.hive.engine.stats["latency_ms"][-10:]))) if self.hive.engine.stats["latency_ms"] else 0
            if lat>1200: ovr["retrieval_k"]=max(3,self.hive.retrieval_k-1)

            if ovr:
                ok,_=self.hive.overlay.patch(ovr, actor_role="hive")
                if ok: self.hive.overlay.apply_to(self.hive)

            if CFG["ALLOW_SELF_REBOOT"] and vm.percent>94:
                safe_reboot("refresh memory")

from abc import ABC, abstractmethod # type: ignore


class IModule(ABC): # type: ignore
    """Interface for a Hive module."""
    def __init__(self, hive_instance: "Hive"):
        self.hive = hive_instance

    @abstractmethod
    def start(self):
        """Start the module."""
        pass

    @abstractmethod
    def stop(self):
        """Stop the module."""
        pass

    def get_status(self) -> dict:
        return {"status": "unknown"}

class ModuleManager:
    """Manages the lifecycle of Hive modules."""
    def __init__(self):
        self.modules: "OrderedDict[str, IModule]" = collections.OrderedDict()

    def register(self, name: str, module: IModule):
        self.modules[name] = module

    def start_all(self):
        print("[ModuleManager] Starting all modules...")
        for name, module in self.modules.items():
            print(f"[ModuleManager] Starting {name}...")
            module.start()
        print("[ModuleManager] All modules started.")

    def stop_all(self):
        print("[ModuleManager] Stopping all modules...")
        for name, module in reversed(self.modules.items()):
            module.stop()
        print("[ModuleManager] All modules stopped.")

# ----------- internal optimization stack -----------
def _append_jsonl(path, rec):
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")

@dataclass
class ChangeProposal: # type: ignore
    kind: str           # "model" | "package" | "code"
    name: str           # model id / package name / file target
    version: str = "" # type: ignore
    patch_text: str = "" # for "code": full replacement or diff
    reason: str = "" # type: ignore
    created_ts: float = field(default_factory=time.time)
    proposer: str = "hive" # type: ignore
    id: str = "" # type: ignore

class Sandbox:
    def __init__(self):
        self.root=os.path.join(OPT_DIR, f"sandbox_{int(time.time())}")
        os.makedirs(self.root, exist_ok=True)
        self.venv=os.path.join(self.root,"venv")
    def _run(self, args, timeout):
        p=subprocess.run(args, capture_output=True, text=True, timeout=timeout)
        return p.returncode, (p.stdout or "") + (p.stderr or "")
    def create(self):
        rc,out=self._run([sys.executable,"-m","venv",self.venv], timeout=120)
        if rc!=0: raise RuntimeError("venv create failed: "+out)
    def pip(self, pkg_spec):
        py=os.path.join(self.venv,"bin","python") if os.name!="nt" else os.path.join(self.venv,"Scripts","python.exe")
        rc,out=self._run([py,"-m","pip","install","--upgrade",pkg_spec], timeout=CFG["OPT_SANDBOX_TIMEOUT"])
        if rc!=0: raise RuntimeError("pip install failed: "+out)
    def run_snippet(self, code:str):
        py=os.path.join(self.venv,"bin","python") if os.name!="nt" else os.path.join(self.venv,"Scripts","python.exe")
        tmp=os.path.join(self.root,"snippet.py"); open(tmp,"w",encoding="utf-8").write(code)
        rc,out=self._run([py,tmp], timeout=CFG["OPT_SANDBOX_TIMEOUT"]); return rc,out

def _synthetic_eval(hive_factory, prompts: List[str]) -> Dict:
    lat_ms=[]; toks_s=[]; quality=0.0
    for p in prompts:
        t0=time.time()
        h=hive_factory()
        out=h.pipe(h.compiler.compile(p, []), max_new_tokens=64, do_sample=False, temperature=0.2) # type: ignore
        t1=time.time()
        text=out[0]["generated_text"]
        lat_ms.append((t1-t0)*1000)
        toks=max(1,len(text.split())); toks_s.append(toks/max(0.001,(t1-t0)))
        q=sum(1 for w in set(re.findall(r"\w+", p.lower())) if w in text.lower())/max(1,len(set(re.findall(r"\w+", p.lower()))))
        quality+=q
    n=max(1,len(prompts))
    return {"lat_ms":sum(lat_ms)/n, "toks_s":sum(toks_s)/n, "quality":quality/n}

class ChangeManager:
    def __init__(self, hive_cls):
        self.hive_cls=hive_cls
    def _allowed_pkg(self, name): 
        return any(name.strip().startswith(allow.strip()) for allow in CFG["OPT_PKG_ALLOWLIST"])
    def _allowed_model(self, mid):
        return mid in CFG["OPT_MODEL_ALLOWLIST"]
    def propose(self, cp: ChangeProposal)->str:
        cp.id=f"chg_{int(time.time())}_{abs(hash(cp.name))%100000}"; _append_jsonl(OPT_PROPOSALS, cp.__dict__); return cp.id
    def test_and_compare(self, cp_id:str, proposal: ChangeProposal)->Dict:
        """
        Tests a proposal in a sandbox, compares it against the baseline,
        and automatically applies it if it passes and auto-apply is enabled.
        """
        def base_hive(): return self.hive_cls(model_id=None, lite=True)
        prompts=["Summarize the water cycle.","Translate to French: the quick brown fox jumps over the lazy dog.","Two-sentence difference between TCP and UDP."]
        base=_synthetic_eval(base_hive, prompts)
        sand=Sandbox(); sand.create()
        model_override=None
        try:
            # Install requirements in sandbox venv
            reqs = ["numpy>=1.24.0","psutil>=5.9.0","requests>=2.31.0","gradio>=4.44.0","sentence-transformers>=3.0.0","faiss-cpu>=1.8.0",
                    "transformers>=4.44.0","accelerate>=0.33.0","datasets>=2.21.0","soundfile>=0.12.1","faster-whisper>=1.0.0","langid>=1.1.6",
                    "piper-tts>=1.2.0","g2p_en>=2.1.0","librosa>=0.10.1","scikit-learn>=1.1.0","feedparser>=6.0.11","duckduckgo_search>=6.2.10",
                    "keyring>=24.3.1"]
            for req in reqs:
                sand.pip(req)

            if proposal.kind=="package":
                if not self._allowed_pkg(proposal.name): return {"ok":False,"reason":"package not allowlisted"}
                spec=proposal.name + (("=="+proposal.version) if proposal.version else "")
                sand.pip(spec)
            elif proposal.kind=="model":
                if not self._allowed_model(proposal.name): return {"ok":False,"reason":"model not allowlisted"}
                model_override=proposal.name
            elif proposal.kind=="code":
                target=os.path.basename(__file__); patched=os.path.join(sand.root,target)
                with open(patched,"w",encoding="utf-8") as f: f.write(proposal.patch_text or "")
                code=f"import importlib.util, json; p=r'{patched}'; spec=importlib.util.spec_from_file_location('hmod',p); m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m); h=m.Hive(); print(json.dumps({{'ok':True}}))"
                rc,out=sand.run_snippet(code)
                if rc!=0 or '"ok": true' not in out.lower(): return {"ok":False,"reason":"patch smoke test failed","out":out}
        except Exception as e:
            return {"ok":False,"reason":f"sandbox setup failed: {e}"}
        def cand_hive(): return self.hive_cls(model_id=model_override, lite=True) if model_override else self.hive_cls(model_id=None, lite=True)
        cand=_synthetic_eval(cand_hive, prompts)
        delta={"lat_ms": base["lat_ms"]-cand["lat_ms"], "toks_s": cand["toks_s"]-base["toks_s"], "quality": cand["quality"]-base["quality"]}
        passed=True
        if CFG["OPT_THRESH_LATENCY_MS"]>0 and delta["lat_ms"]<CFG["OPT_THRESH_LATENCY_MS"]: passed=False
        if CFG["OPT_THRESH_TOKS_PER_S"]>0 and delta["toks_s"]<CFG["OPT_THRESH_TOKS_PER_S"]: passed=False
        if delta["quality"]<CFG["OPT_THRESH_QUALITY"]: passed=False
        result={"ok":True,"proposal":proposal.__dict__,"base":base,"cand":cand,"delta":delta,"passed":passed, "ts": time.time()}
        _append_jsonl(OPT_RESULTS, result)

        # Automatically apply if tests passed and auto-apply is on
        if passed and CFG.get("OPT_AUTO_APPLY"):
            apply_ok, apply_msg = self.apply(result)
            result["applied"] = {"ok": apply_ok, "message": apply_msg, "ts": time.time()}
            _append_jsonl(OPT_RESULTS, {"update_for": cp_id, "applied": result["applied"]})
        return result
    def apply(self, result:Dict)->Tuple[bool,str]:
        prop=result.get("proposal",{}); kind=prop.get("kind"); name=prop.get("name","")
        if not result.get("passed"): return False,"did not meet thresholds"
        if kind=="package":
            if not self._allowed_pkg(name): return False,"package not allowlisted"
            try:
                subprocess.check_call([sys.executable,"-m","pip","install","--upgrade", name + (("=="+prop.get("version","")) if prop.get("version") else "")])
                return True,"package installed"
            except Exception as e: return False,f"pip failed: {e}"
        if kind=="model":
            if not self._allowed_model(name): return False,"model not allowlisted"
            pref=os.path.join(OPT_DIR,"preferred_model.json"); _atomic_write_json(pref, {"model_id":name,"ts":time.time()})
            return True,"model preference recorded (takes effect after restart)"
        if kind=="code":
            is_pi = 'raspberrypi' in platform.machine().lower()
            if is_pi and hasattr(self.hive_cls, 'bootstrap_instance') and self.hive_cls.bootstrap_instance:
                print("[ChangeManager] Raspberry Pi detected, attempting hot-reload.")
                try:
                    target=os.path.abspath(__file__)
                    with open(target, "w", encoding="utf-8") as f: f.write(prop.get("patch_text","") or "")
                    self.hive_cls.bootstrap_instance.soft_restart()
                    return True, "Code hot-reloaded without a full reboot."
                except Exception as e:
                    return False, f"Hot-reload failed: {e}. A manual restart is required."

            try:
                target=os.path.abspath(__file__); backup=target+f".bak_{int(time.time())}"; shutil.copyfile(target,backup)
                with open(target,"w",encoding="utf-8") as f: f.write(prop.get("patch_text","") or ""); return True,"code updated (backup created); restart recommended"
            except Exception as e: return False,f"code write failed: {e}"
        return False,"unknown change type"

class ChangeManagerModule(ChangeManager, IModule): # type: ignore
    def __init__(self, hive_instance: "Hive"):
        IModule.__init__(self, hive_instance)
        ChangeManager.__init__(self, hive_instance.__class__)
    
    def start(self): pass
    def stop(self): pass

class SelfOptimizerModule(SelfOptimizer, IModule):
    def __init__(self, hive_instance: "Hive"):
        IModule.__init__(self, hive_instance)
        SelfOptimizer.__init__(self, hive_instance)
    
    def start(self):
        super().start()
    def stop(self): self.stop = True

class LibrarianCurve:
    """Implements the Librarian from Part 2, Section 7."""
    def __init__(self, curve_store: CurveStore, k_store: KnowledgeStore):
        self.store = curve_store
        self.k_store = k_store

    def retrieve_scoped_with_scores(self, query: str, role: str, user_id: Optional[str], k: int = 6):
        # This is a simplified retrieval. A full implementation would use the role and user_id for scoping.
        return self.store.search_with_scores(query, k=k)

class VoiceServicesModule(IModule):
    def __init__(self, hive_instance: "Hive"):
        super().__init__(hive_instance)

    def start(self):
        if _HAVE_VAD:
            self.hive.vad_service = VADService(aggressiveness=CFG["VOICE_VAD_AGGRESSIVENESS"])
        self.hive.asr_service = ASRService()
        self.hive.tts_service = TTSService()
        self.hive.video_service = VideoService(self.hive)
        if self.hive.video_service: self.hive.video_service.start()

    def stop(self):
        if self.hive.video_service: self.hive.video_service.stop_event.set()

class VideoService(IModule):
    """Handles video capture from a webcam."""
    def __init__(self, hive_instance: "Hive"):
        super().__init__(hive_instance)
        self.cap = None
        if _HAVE_CV:
            # Initialize the camera capture
            self.cap = cv2.VideoCapture(0)

    def get_frame(self):
        if not self.cap: return None
        ret, frame = self.cap.read()
        return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) if ret else None

class PersistenceEngine(IModule):
    """Placeholder for a module that would handle data persistence strategies."""
    def __init__(self, hive_instance: "Hive"):
        super().__init__(hive_instance)

    def start(self): pass
    def stop(self): pass

# ----------- Hive core -----------

# type: ignore
class PromptCompiler:
    def __init__(self):
        self.override_head=None
        self.override_budget=None
        self.personas = {
            "default": "You are a helpful assistant. Use the provided facts to answer the user's question concisely.",
            "en": "You are an encouraging and patient English tutor. Use the facts to explain the topic clearly and simply.",
            "essay_review": "You are a writing critic. Provide a detailed review of the following essay, focusing on structure, clarity, and vocabulary. Use the provided facts for context if needed.",
            "pronounce": "You are a pronunciation coach. Explain how to say the word, using the provided phonetic hints.", # type: ignore
        }

    def compile(self, final_instruction: str, snippets: List[Dict], token_budget: int = 600, intent: str = "default", user_prefs: Optional[Dict] = None, role: str = "guest") -> str:
        if self.override_budget: token_budget = self.override_budget
        prefs = user_prefs or {}
        user_lang = prefs.get("language", "en")
        learning_level = prefs.get("learning_level", "intermediate") # e.g., beginner, intermediate, advanced
        
        # Simple ranker: prioritize snippets with more overlapping words.
        query_words = set(re.findall(r"\w+", final_instruction.lower()))
        def rank_score(snippet): # type: ignore
            text = (snippet.get("text", "") or "").lower()
            return len(query_words.intersection(re.findall(r"\w+", text)))
        ranked = sorted(snippets, key=rank_score, reverse=True)
        
        # Synthesize a concise "insight" from the best snippets instead of just listing them.
        # This creates a more natural and integrated prompt for the LLM.
        insight = ""
        if ranked:
            top_snippet_text = (ranked[0].get("text", "") or "").strip()
            # Create a very short, focused summary of the most relevant fact.
            insight_summary = ' '.join(top_snippet_text.split()[:25]) + ('...' if len(top_snippet_text.split()) > 25 else '')
            insight = f"Based on my knowledge, I know that: \"{insight_summary}\". Use this key insight to inform your answer."

        # Select persona based on intent and user profile
        head = self.override_head or self.personas.get(intent, self.personas.get(user_lang, self.personas["default"]))
        
        # Add personalization based on user profile
        if learning_level == "beginner":
            head += " Keep your language very simple and be extra encouraging."
        if role in ("owner", "admin_super", "admin_general"):
            head += f" You are speaking to an administrator ({role}). You may provide more technical details or system status if relevant."
        
        return f"{head} {insight}\n\nUser: {final_instruction}\nAssistant:"

class KnowledgeStoreModule(KnowledgeStore, IModule): # type: ignore
    def __init__(self, hive_instance: "Hive"): IModule.__init__(self, hive_instance); KnowledgeStore.__init__(self, hive_instance.config["HIVE_HOME"])
    def start(self): pass
    def stop(self): pass

class CurveStoreModule(CurveStore, IModule): # type: ignore
    def __init__(self, hive_instance: "Hive"):
        IModule.__init__(self, hive_instance)
        CurveStore.__init__(self, hive_instance.config["CURVE_DIR"])
    def start(self): pass
    def stop(self): pass

class EngineModule(EngineCurve, IModule):
    def __init__(self, hive_instance: "Hive"):
        IModule.__init__(self, hive_instance)
        EngineCurve.__init__(self)
    def start(self): pass
    def stop(self): pass
    
class OverlayModule(RuntimeOverlay, IModule):
    def __init__(self, hive_instance: "Hive"):
        IModule.__init__(self, hive_instance)
        RuntimeOverlay.__init__(self)
    def start(self): self.apply_to(self.hive)
    def stop(self): pass

class CompilerModule(PromptCompiler, IModule):
    def __init__(self, hive_instance: "Hive"): IModule.__init__(self, hive_instance); PromptCompiler.__init__(self); hive_instance.decoding_temperature=0.7
    def start(self): pass
    def stop(self): pass

class Hive:
    def __init__(self, model_id: Optional[str]=None, device: Optional[str]=None, caps: Optional[Dict]=None, lite: bool = False):
        self.config = CFG
        self.caps = caps or probe_caps()
        self.lite_mode = lite
        self.module_manager = ModuleManager() # type: ignore
        Hive.bootstrap_instance = None # Class attribute to hold bootstrap instance
        self.llm_ready = threading.Event()
        self.pipe = None
        self.tok = None
        self.model = None

        if not model_id:
            model_id, info = pick_model(self.caps)
            device = info.get("device", "cpu")
        self.model_id = model_id or CFG["MODEL_OVERRIDE"] or CANDIDATES[0][0]
        self.device = device or ("cuda" if _has_gpu_env() else "cpu")
        
        if self.lite_mode:
            self._init_lite_mode()
        else:
            self._init_full_mode()

    def _init_lite_mode(self): # type: ignore
        """Initializes the Hive in lite mode."""
        print("[Hive] Initializing in Lite Mode.")
        self._setup_llm_pipeline()

    def _init_full_mode(self):
        """Initializes the Hive in full-featured mode."""
        print("[Hive] Initializing in Full Mode.")
        self.module_manager.register("kstore", KnowledgeStoreModule(self))
        self.module_manager.register("store", CurveStoreModule(self))
        self.module_manager.register("librarian", LibrarianModule(self))
        self.module_manager.register("compiler", CompilerModule(self))
        self.module_manager.register("engine", EngineModule(self))
        self.module_manager.register("overlay", OverlayModule(self))
        self.module_manager.register("changes", ChangeManagerModule(self))
        self.module_manager.register("voice_video", VoiceServicesModule(self))
        self.module_manager.register("persistence", PersistenceEngine(self))
        self.module_manager.register("selfopt", SelfOptimizerModule(self))
        self.module_manager.register("dialogue", DialogueManager(self))
        self._setup_llm_pipeline()
        self.module_manager.start_all()

    def _load_local_model(self, trust: bool, **kwargs):
        """Loads the tokenizer and model for local inference."""
        print(f"[Hive] Loading local model: {self.model_id} on device: {self.device}")
        self.tok = AutoTokenizer.from_pretrained(self.model_id, trust_remote_code=trust, chat_template=None)
        if self.tok.pad_token is None:
            self.tok.pad_token = self.tok.eos_token
        
        self.model = AutoModelForCausalLM.from_pretrained(self.model_id, trust_remote_code=trust, **kwargs)
        self.model.eval()

        # Define stop tokens for generation
        stop_token_names = ["<|endoftext|>", "<|file_separator|>", "<|user|>", "<|assistant|>", "<|im_start|>", "<|im_end|>", "</s>"] # type: ignore
        self.stop_tokens = [tid for tid in self.tok.convert_tokens_to_ids(stop_token_names) if tid is not None]
        if self.tok.eos_token_id is not None:
            self.stop_tokens.append(self.tok.eos_token_id)
        self.stopping_criteria = StoppingCriteriaList([StopOnTokens(self.stop_tokens)])

    def _setup_llm_pipeline(self):
        """Sets up the language model, tokenizer, and pipeline."""
        trust = True; kwargs = {}
        if torch and torch.cuda.is_available() and self.device == "cuda":
            kwargs.update(dict(torch_dtype=torch.float16, device_map="auto"))

        # --- Automatic Inference Mode Switching ---
        # Default to local inference for Pi/local machines, remote for HF Spaces.
        # This can be manually overridden by setting HIVE_USE_HF_INFERENCE.
        is_hf_space = "SPACE_ID" in os.environ
        use_remote_default = is_hf_space
        print(f"[Hive] Detected Hugging Face Space: {is_hf_space}. Defaulting to remote inference: {use_remote_default}.")
        # Check for manual override from environment variable
        if "HIVE_USE_HF_INFERENCE" in os.environ:
            use_remote = CFG["HIVE_USE_HF_INFERENCE"]
        else:
            use_remote = use_remote_default

        if use_remote:
            print("[Hive] Using remote Hugging Face Inference endpoint.", flush=True)
            from huggingface_hub import InferenceClient; endpoint = CFG["HIVE_HF_ENDPOINT"] or None; token = CFG["HF_READ_TOKEN"] or os.getenv("HF_TOKEN") or os.getenv("HUGGING_FACE_HUB_TOKEN") or None
            self.client = InferenceClient(model=self.model_id if endpoint is None else None, token=token, timeout=60, base_url=endpoint) # type: ignore
            def _remote_pipe(prompt, max_new_tokens=256, do_sample=True, temperature=0.7, **kw):
                messages = [{"role": "user", "content": prompt}]
                resp = self.client.chat_completion(messages, max_tokens=int(max_new_tokens), temperature=float(temperature), do_sample=bool(do_sample), stream=False)
                return [{"generated_text": resp.choices[0].message.content}]
            self.pipe = _remote_pipe
            # For remote inference, we still need a local tokenizer for prompt compilation
            self.tok = AutoTokenizer.from_pretrained(self.model_id, trust_remote_code=trust, chat_template=None) # type: ignore
            # We pass `token=False` to prevent from_pretrained from using a potentially invalid # type: ignore
            # environment token, as we only need the public tokenizer config.
            # The actual inference call uses the token provided to the InferenceClient.
            self.tok = AutoTokenizer.from_pretrained(self.model_id, trust_remote_code=trust, chat_template=None, token=False)
            self.model = None # No local model needed
            self.stopping_criteria = None # Not used with InferenceClient
        else:
            print("[Hive] Using local LLM for inference.", flush=True)
            self.tok = AutoTokenizer.from_pretrained(self.model_id, trust_remote_code=trust, chat_template=None)
            if self.tok.pad_token is None:
                self.tok.pad_token = self.tok.eos_token
            self.model = AutoModelForCausalLM.from_pretrained(self.model_id, trust_remote_code=trust, **kwargs)
            
            self.model.eval()
            
            self.stop_tokens = self.tok.convert_tokens_to_ids(["<|endoftext|>", "<|file_separator|>","<|user|>","<|assistant|>","<|im_start|>","<|im_end|>","</s>"])
            self.stop_tokens.append(self.tok.eos_token_id)
            self.stopping_criteria = StoppingCriteriaList([StopOnTokens(self.stop_tokens)])
            # The pipeline object does not support streaming well with StoppingCriteria. We will call the model directly for streaming.
            self.pipe = pipeline("text-generation", model=self.model, tokenizer=self.tok, device=self.device, stopping_criteria=self.stopping_criteria)
        self.llm_ready.set()

    @property
    def store(self) -> 'CurveStore': return self.module_manager.modules["store"] # type: ignore
    @property
    def librarian(self) -> 'LibrarianCurve': return self.module_manager.modules["librarian"] # type: ignore
    @property
    def engine(self) -> 'EngineCurve': return self.module_manager.modules["engine"] # type: ignore
    @property
    def overlay(self) -> 'RuntimeOverlay': return self.module_manager.modules["overlay"] # type: ignore
    @property
    def changes(self) -> 'ChangeManager': return self.module_manager.modules["changes"] # type: ignore
    @property
    def compiler(self) -> 'PromptCompiler': return self.module_manager.modules["compiler"] # type: ignore
    @property
    def selfopt(self) -> 'SelfOptimizer': return self.module_manager.modules["selfopt"] # type: ignore

    @property
    def persistence(self) -> 'PersistenceEngine': return self.module_manager.modules["persistence"] # type: ignore
    @property
    def dialogue_manager(self) -> 'DialogueManager': return self.module_manager.modules["dialogue"] # type: ignore
    def _prepare_chat_input(self, message: str, user_lang: str, phonics_on: bool, prompt_override: str | None) -> tuple[str, str]: # type: ignore
        """Determines intent and prepares the final message for the LLM."""
        intent = self.engine.choose_route(message)
        final_message = message

        if intent == "pronounce" or (phonics_on and user_lang == 'en'):
            match = re.search(r"(pronounce|say|spell|spelling of)\s+['\"]?([a-zA-Z\-']+)['\"]?", message, re.I)
            word_to_process = match.group(2) if match else (message.split()[-1] if len(message.split()) < 4 else None)
            if word_to_process:
                phonics_hint = phonics(word_to_process)
                final_message = f"Explain how to pronounce the word '{word_to_process}'. Use this phonics hint in your explanation: {phonics_hint}"
        elif prompt_override:
            final_message = f"{prompt_override}\n\nHere is the text to work on:\n{message}"
            if "review" in prompt_override.lower() or "essay" in prompt_override.lower():
                intent = "essay_review"

        return final_message, intent

    def _get_retrieval_context(self, message: str, effective_role: str, caller_id: str | None, k: int) -> list[dict]: # type: ignore
        """Performs RAG, with web search fallback if necessary."""
        if self.lite_mode:
            return []

        online_now = NET.online_quick()
        if not online_now:
            NET.kick_async()

        snippets, scores = self.librarian.retrieve_scoped_with_scores(message, effective_role, caller_id, k=k)
        cov = coverage_score_from_snippets(snippets, scores) # type: ignore

        if cov < self.web_threshold and CFG["ONLINE_ENABLE"] and online_now:
            self.web_update_and_store(message, max_docs=int(CFG["ONLINE_MAX_RESULTS"] or 5), timeout=int(CFG["ONLINE_TIMEOUT"] or 8))
            snippets, _ = self.librarian.retrieve_scoped_with_scores(message, effective_role, caller_id, k=k)

        return snippets
    
    def _postprocess_and_log(self, full_output: str, message: str, effective_role: str, caller_id: str | None, intent: str, snippets: list[dict]):
        """Cleans the LLM output and logs the interaction."""
        reply = full_output.rsplit("Assistant:", 1)[-1].strip()
        if CFG["NO_PROFANITY"]:
            reply = re.sub(r"\b(fuck|shit|bitch|asshole|cunt|dick|pussy|nigger|motherfucker)\b", "[censored]", reply, flags=re.I)

        if caller_id and not self.lite_mode:
            log_path = os.path.join(CFG["HIVE_HOME"], "users", "conversations", f"{caller_id}.jsonl")
            log_entry = {"ts": time.time(), "message": message, "effective_role": effective_role, "intent": intent, "snippets_used": [s.get("text", "")[:100] for s in snippets[:3]], "reply": reply}
            _append_jsonl(log_path, log_entry)

        return reply

    def summarize_for_memory(self, text:str, max_new_tokens:int=160)->str:
        prompt=("Condense the following content into 4–6 bullet points with names, dates, numbers, and a one-line takeaway. Keep it factual.\n\n"
                f"{text[:3000]}\n\nSummary:")
        out=self.pipe(prompt, max_new_tokens=max_new_tokens, do_sample=False, temperature=0.01)
        return out[0]["generated_text"].split("Summary:",1)[-1].strip()

    def add_curve(self, text:str, meta:Dict, scope:str="general"): # type: ignore
        if self.lite_mode: return
        self.librarian.ingest_text(text, meta, scope)

    def online_update(self, query_hint: Optional[str]=None)->Dict:
        if self.lite_mode: return {"ok": False, "reason": "Online features are disabled in Lite Mode."}

        if not CFG["ONLINE_ENABLE"]: return {"ok":False,"reason":"online disabled"}
        if not online_available(int(CFG["ONLINE_TIMEOUT"])): return {"ok":False,"reason":"offline"}
        seen=_load_json(ONLINE_DB, {}) # type: ignore
        urls=[u.strip() for u in (CFG["ONLINE_SOURCES"] or "").split(",") if u.strip()]
        items=fetch_rss(urls, timeout=int(CFG["ONLINE_TIMEOUT"]), limit=30)
        added=0
        for it in items: # type: ignore

            key=hashlib.sha1(((it.get("link") or "")+(it.get("title") or "")).encode("utf-8","ignore")).hexdigest()
            if key in seen: continue

            base=(it.get("title","")+"\n\n"+it.get("summary","")).strip()
            summ=self.summarize_for_memory(base)
            self.add_curve(summ, {"dataset":"online_rss","url":it.get("link"),"title":it.get("title"),"published":it.get("published")}, scope="general")

            seen[key]=int(time.time()); added+=1 # type: ignore
        _save_json(ONLINE_DB, seen); return {"ok":True,"added":added}

    def web_update_and_store(self, query:str, max_docs:int, timeout:int)->int:
        if self.lite_mode: return 0 # type: ignore
        if not (CFG["ONLINE_ENABLE"] and online_available(timeout)): return 0
        hits=asyncio.run(web_search_snippets(query, max_results=max_docs, timeout=timeout)); added=0
        for h in hits:
            body=(h.get("title","")+"\n\n"+(h.get("body","") or "")).strip()
            if not body: continue
            summ=self.summarize_for_memory(body)
            meta={"dataset":"web_update","source":h.get("href",""),"title":h.get("title",""),"ts":time.time()}
            self.add_curve(summ, meta, scope="general"); added+=1
        return added

    def chat_stream(self, prompt: str, max_new_tokens: int, temperature: float):
        """Generator that yields tokens as they are generated."""
        if hasattr(self, 'client') and self.client: # Remote Inference
            stop_sequences = ["</s>", "Assistant:"] + [self.tok.decode(st) for st in self.stop_tokens]
            try:
                messages = [{"role": "user", "content": prompt}]
                for chunk in self.client.chat_completion(
                    messages=messages, max_tokens=int(max_new_tokens), temperature=float(temperature), 
                    do_sample=True, stop=stop_sequences, stream=True
                ):
                    content = chunk.choices[0].delta.content
                    if content:
                        yield content
            except Exception as e:
                print(f"[ModelBridge] Remote inference stream failed: {e}")
                yield "[Error: Could not get response from remote model]"
            return

        if not (hasattr(self, 'model') and self.model): # Local model not loaded
            yield "[Error: Local model is not available]"
            return
        
        streamer = TextIteratorStreamer(self.tok, skip_prompt=True, skip_special_tokens=True)
        inputs = self.tok([prompt], return_tensors="pt").to(self.device) # type: ignore
        generation_kwargs = dict(
            inputs,
            streamer=streamer, # type: ignore
            max_new_tokens=max_new_tokens,
            do_sample=True,
            temperature=temperature,
            stopping_criteria=self.stopping_criteria
        )
        thread = threading.Thread(target=self.model.generate, kwargs=generation_kwargs)
        thread.start()
        for new_text in streamer:
            yield new_text

    def chat(self, message:str, effective_role:str, caller_id: Optional[str],
             k:int=None, max_new_tokens:int=1024, temperature:float=None, prompt_override: Optional[str] = None) -> str: # type: ignore
        temp = temperature if temperature is not None else (self.decoding_temperature if not self.lite_mode else 0.7)
        
        # This logic was previously in _prepare_chat_input
        user_prefs = self.dialogue_manager.get_user_prefs(caller_id) if hasattr(self, 'dialogue_manager') else {}
        final_message, intent = self._prepare_chat_input(message, user_prefs.get("language", "en"), user_prefs.get("phonics_on", False), prompt_override)

        if self.lite_mode:
            prompt = f"<|user|>\n{message}</s>\n<|assistant|>\n"
            full_reply = "".join(list(self.chat_stream(prompt, max_new_tokens=max_new_tokens, temperature=temp)))
            return full_reply
        
        kk = k if k is not None else (self.retrieval_k if hasattr(self, 'retrieval_k') else 6)
        snippets = self._get_retrieval_context(message, effective_role, caller_id, kk) # type: ignore

        prompt = self.compiler.compile( # type: ignore
            final_message, 
            snippets,
            token_budget=int(CFG["CTX_TOKENS"]), 
            intent=intent
        )

        full_output = "".join(list(self.chat_stream(prompt, max_new_tokens, temp))) # type: ignore
        self.engine.run(message, snippets)

        return self._postprocess_and_log(full_output, message, effective_role, caller_id, intent, snippets)

# --------------- UI ---------------
HELP=f"""
**Admin/User mode**: Admins (general/super) and Owner log in with password (Owner also needs second factor). After login choose Admin or User mode.
**Owner-only code edits** are enforced via Change Manager policy. Hive can sandbox, test, and propose; code writes require Owner approval (`OPT_AUTO_APPLY=1`) unless Owner applies manually.
**Offline/Online**: Works fully offline from curves. If online and enabled, fetches RSS/web snippets ➡️ summarizes locally ➡️ saves to curves (persists offline).
**Voice**: Faster-Whisper ASR (auto language), Piper TTS mixed-language, phonics hints (English).
**Privacy**: Sensitive/first-person inputs route to user-private library; neutral info to general.
"""

def launch_ui(bootstrap_instance: "Bootstrap"):
    with gr.Blocks(title="Hive 🐝") as demo:
        with gr.Row():
            with gr.Column(scale=3):
                gr.Markdown(f"## {CFG['AGENT_NAME']} 🐝")
                core_status = gr.Markdown("⏳ **Initializing Full Hive Core...** (Est. 1-5 mins). You can chat with the Lite model now. Advanced features will be enabled shortly.") # type: ignore
                chatbot = gr.Chatbot(height=600, type="messages", label="Chat", placeholder="Initializing...")
                msg = gr.Textbox(placeholder="Please wait for the model to load...", interactive=False, show_label=False, container=False, scale=4)


            with gr.Column(scale=1, min_width=300):
                with gr.Sidebar():
                    uid_state=gr.State(None); role_state=gr.State("guest"); mode_state=gr.State("user"); phonics_state=gr.State(False) # type: ignore

                    with gr.Accordion("Login & Profile", open=True):
                        login_name=gr.Textbox(label="Name or ID")
                        login_pass=gr.Textbox(label="Password (if required)", type="password")
                        login_second=gr.Textbox(label="Second (owner only)", type="password")
                        login_btn=gr.Button("Login")
                        login_status=gr.Markdown(elem_id="login_status") # type: ignore
                        profile_status = gr.Markdown("Login to see your profile.")
                        profile_save_btn = gr.Button("Save Profile")

                    with gr.Accordion("🌐 Language Preference", open=False):
                        profile_lang = gr.Dropdown(choices=["en","es","fr","de","zh"], label="Preferred Language", value="en")

                    with gr.Accordion("🗣️ Phonics Assist", open=False):
                        gr.Markdown("Enable to get phonetic hints for English words when using the 'pronounce' command.")
                        profile_phonics = gr.Checkbox(label="Enable Phonics Assist (for English)")

                    with gr.Accordion("🧠 Memory & Vocabulary", open=False):
                        summary_output = gr.Markdown("Initializing... (Full core required, est. 1-2 min)")
                        summary_btn = gr.Button("Show Memory Summary", interactive=False)
                        vocab_output = gr.Markdown("---")
                        vocab_btn = gr.Button("Get New Word", interactive=False)
                        progress_output = gr.Markdown("---")

                    with gr.Accordion("🗣️ Voice & Hands-Free", open=False, visible=True) as voice_accordion:
                        voice_status_md = gr.Markdown("Initializing voice models... (Est. 15-90 sec)")
                        with gr.Tabs() as voice_tabs:
                            with gr.TabItem("Push-to-Talk"):
                                ptt_audio_in = gr.Audio(sources=["microphone"], type="filepath", label="1. Record your message", interactive=False)
                                ptt_transcript = gr.Textbox(label="2. Transcript / Your Message", interactive=False)
                                with gr.Row():
                                    ptt_transcribe_btn = gr.Button("Transcribe Only", interactive=False)
                                    ptt_chat_btn = gr.Button("Send to Chat & Get Voice Reply", variant="primary", interactive=False)
                                ptt_reply_audio = gr.Audio(type="filepath", label="3. Assistant's Voice Reply", autoplay=True)
                            with gr.TabItem("Hands-Free"):
                                vocal_chat_state = gr.State({"active": False, "audio_buffer": b'', "last_interaction_time": 0, "conversation_timeout": 10.0})
                                vocal_chat_btn = gr.Button("Start Hands-Free Conversation", interactive=False)
                                vocal_chat_status = gr.Markdown("Status: Inactive")
                                vocal_mic = gr.Audio(sources=["microphone"], streaming=True, visible=False, autoplay=True)
                                wake_word_mic = gr.Audio(sources=["microphone"], streaming=True, visible=False, autoplay=False, elem_id="wake_word_mic")
                                wake_word_state = gr.State({"buffer": b""})
                            with gr.TabItem("Voice Login"):
                                gr.Markdown("Enroll your voice to enable password-free login for user accounts.")
                                enroll_audio = gr.Audio(sources=["microphone"], type="filepath", label="Record 5-10s for voiceprint", interactive=False)
                                with gr.Row():
                                    enroll_btn = gr.Button("Enroll Voice for Current User", interactive=False)
                                    enroll_status = gr.Markdown()
                                gr.Markdown("---")
                                gr.Markdown("After enrolling, you can log in by recording your voice here.")
                                with gr.Row():
                                    who_btn = gr.Button("Login by Voice", interactive=False)
                                    who_status = gr.Markdown()

                    with gr.Accordion("📸 Camera", open=False, visible=True) as camera_accordion:
                        camera_status_md = gr.Markdown("Camera feature disabled or initializing...")
                        video_out = gr.Image(label="Camera", type="pil", interactive=False)

                    with gr.Accordion("🌐 Network", open=False, visible=True) as network_accordion:
                        network_status_md = gr.Markdown("Initializing network features...")
                        wifi_status=gr.Markdown("Wi-Fi: checking...")
                        connect_now=gr.Button("Try auto-connect now (non-blocking)")
                        online_now=gr.Button("Fetch updates now", interactive=False)
                        online_status=gr.Markdown()

                    with gr.Accordion("⚙️ Admin Console", open=False, visible=True) as admin_accordion:
                        admin_info=gr.Markdown("Login as an admin and switch to Admin mode to use these tools.")
                        mode_picker=gr.Radio(choices=["user","admin"], value="user", label="Mode (admins only)")
                        with gr.Tabs() as admin_tabs:
                            with gr.TabItem("User Management"):
                                target=gr.Textbox(label="Target name or id")
                                new_name=gr.Textbox(label="New name")
                                rename_btn=gr.Button("Rename")
                                new_pass=gr.Textbox(label="New password")
                                pass_btn=gr.Button("Change password")
                                new_role=gr.Dropdown(choices=["owner","admin_super","admin_general","user"], value="user", label="New role")
                                role_btn=gr.Button("Change role", elem_id="role_btn")
                                out=gr.Markdown()
                            with gr.TabItem("Add User"):
                                add_name=gr.Textbox(label="Add: name")
                                add_role=gr.Dropdown(choices=["admin_super","admin_general","user"], value="user", label="Add role")
                                add_pass=gr.Textbox(label="Add password (admins only)")
                                add_btn=gr.Button("Add user/admin")
                                out_add=gr.Markdown()
                            with gr.TabItem("System"):
                                ingest_status = gr.Markdown("Memory Ingestion: Idle")
                                ingest_now_btn = gr.Button("Start Background Ingestion", interactive=False)
                                mem_compress_btn=gr.Button("Compress Memory (archive)", interactive=False)
                                compress_status=gr.Markdown("")
                                hotpatch_patch=gr.Code(label="Paste hotpatch JSON (advanced)")
                                hotpatch_status=gr.Markdown("Awaiting patch")
                                hotpatch_apply=gr.Button("Apply Hotpatch", elem_id="hotpatch_apply", interactive=False)
                            with gr.TabItem("Optimization"):
                                gr.Markdown("### Internal Optimization (Change Manager)")
                                prop_kind=gr.Dropdown(choices=["model","package","code"], value="model", label="Proposal type")
                                prop_name=gr.Textbox(label="Model ID / Package Name")
                                prop_ver=gr.Textbox(label="Package version (optional)")
                                prop_reason=gr.Textbox(label="Why this change?")
                                prop_patch=gr.Code(label="Code patch (for 'code' proposals): paste full replacement or diff")
                                propose_btn=gr.Button("Propose", interactive=False)
                                test_btn=gr.Button("Test in sandbox", interactive=False)
                                apply_btn=gr.Button("Apply (policy-checked)", elem_id="apply_btn", interactive=False)
                                opt_out=gr.JSON(label="Result")

        # --- Event Handlers --- 

        def _sanitize_input(text: str) -> str:
            """Removes control characters and leading/trailing whitespace."""
            if not text: return ""
            return "".join(ch for ch in text if unicodedata.category(ch)[0] != "C").strip()
        
        def talk(m, uid, role, mode, hist, request: gr.Request): # type: ignore
            effective_role = role if mode == "admin" else "user"
            session_id = request.session_hash
            # Use session_id for guests, uid for logged-in users
            current_user_id = uid or session_id
            
            sanitized_m = _sanitize_input(m)
            if not sanitized_m:
                yield hist, gr.Textbox()
                return
            
            current_history = (hist or []) + [{"role": "user", "content": sanitized_m}]
            yield current_history, gr.Textbox(value="", interactive=False) # Show user message, disable textbox

            hive_instance = get_hive_instance(bootstrap_instance) # type: ignore
            
            if hive_instance.lite_mode:
                # Lite mode: direct, non-streaming response.
                reply = hive_instance.chat(sanitized_m, effective_role, current_user_id)
                current_history.append({"role": "assistant", "content": reply or "[No response from model]"})
                yield current_history, gr.Textbox(value="", interactive=True)
            else:
                # Full mode uses the DialogueManager for a streaming response.
                if not hasattr(hive_instance, 'dialogue_manager'):
                    error_msg = "Dialogue Manager not available. Full core may still be initializing."
                    current_history.append({"role": "assistant", "content": error_msg})
                    yield current_history, gr.Textbox(value="", interactive=True)
                    return
                
                current_history.append({"role": "assistant", "content": ""})
                try:
                    # The dialogue manager needs the full history to maintain context. # type: ignore
                    for chunk in hive_instance.dialogue_manager.process_turn(current_history, current_user_id, effective_role, session_id):
                        if chunk["type"] == "token":
                            current_history[-1]["content"] += chunk["content"]
                            yield current_history, gr.Textbox(value="", interactive=False)
                    # After the stream is complete, re-enable the textbox.
                    yield current_history, gr.Textbox(placeholder=f"Talk to {CFG['AGENT_NAME']}", interactive=True)
                except Exception as e:
                    error_msg = f"Error in DialogueManager: {e}" # type: ignore
                    print(f"[ERROR] {error_msg}")
                    current_history[-1]["content"] = f"An error occurred: {error_msg}"
                    yield current_history, gr.Textbox(value="", interactive=True)

        msg.submit(talk, [msg, uid_state, role_state, mode_state, chatbot], [chatbot, msg], api_name="chat")
        
        def do_memory_summary(uid, request: gr.Request):
            hive_instance = get_hive_instance() # type: ignore
            if hive_instance.lite_mode: return "Memory features are disabled in Lite Mode." # type: ignore
            current_user_id = uid or request.session_hash # type: ignore
            log_path = os.path.join(CFG["HIVE_HOME"], "users", "conversations", f"{current_user_id}.jsonl")
            if not os.path.exists(log_path): return "No conversation history found."
            try: # type: ignore
                with open(log_path, "r", encoding="utf-8") as f:
                    lines = f.readlines()[-10:]
                    if not lines: return "Not enough conversation history to summarize." # type: ignore
                    text_to_summarize = "\n".join([json.loads(line).get("message", "") + "\n" + json.loads(line).get("reply", "") for line in lines])
                    summary = hive_instance.summarize_for_memory(text_to_summarize) # type: ignore
                    return summary if summary.strip() else "Could not generate a summary from recent conversations."
            except Exception as e: return f"Error generating summary: {e}"
        summary_btn.click(do_memory_summary, [uid_state], [summary_output])

        def do_get_vocab_word(uid, request: gr.Request):
            hive_instance = get_hive_instance() # type: ignore
            if hive_instance.lite_mode: return "Vocabulary features are disabled in Lite Mode." # type: ignore
            current_user_id = uid or request.session_hash
            log_path = os.path.join(CFG["HIVE_HOME"], "users", "conversations", f"{current_user_id}.jsonl")
            if not os.path.exists(log_path): return "No conversation history to find words from."
            try:
                with open(log_path, "r", encoding="utf-8") as f:
                    content = f.read()
                words = [w for w in re.findall(r'\b\w{7,}\b', content.lower()) if w not in ["assistant", "message"]]
                if not words: return "No challenging words found yet. Keep chatting!" # type: ignore
                word = random.choice(words)
                definition = hive_instance.chat(f"What is the definition of the word '{word}'? Provide a simple, clear definition and one example sentence.", "user", current_user_id) # type: ignore
                return f"**{word.capitalize()}**: {definition}"
            except Exception as e: return f"Error getting vocabulary word: {e}"

        def wait_for_memory_features():
            """Waits for the full Hive core and enables memory-related UI features."""
            bootstrap_instance.hive_ready.wait() # type: ignore
            hive_instance = get_hive_instance() # Ensure the UI's HIVE_INSTANCE is updated to full
            return (
                "✅ **Full Hive Core is Ready.** Advanced features are now online.",
                "Click the button to generate a summary of your recent conversations.",
                gr.Textbox(placeholder=f"Talk to {CFG['AGENT_NAME']}", interactive=True),
                gr.Button(interactive=True),
                "Click to get a new vocabulary word from your conversations.",
                gr.Button(interactive=True),
                "Your progress will be shown here. Click the button to update.",
                # Enable other advanced feature buttons
                gr.Button(interactive=True), # online_now
                gr.Button(interactive=True), # ingest_now_btn
                gr.Button(interactive=True), # mem_compress_btn
                gr.Button(interactive=True), # hotpatch_apply
                gr.Button(interactive=True), # propose_btn
                gr.Button(interactive=True), # test_btn
                gr.Button(interactive=True), # apply_btn
            )
        demo.load(wait_for_memory_features, None, [core_status, summary_output, msg, summary_btn, vocab_output, vocab_btn, progress_output, online_now, ingest_now_btn, mem_compress_btn, hotpatch_apply, propose_btn, test_btn, apply_btn, network_status_md])
        def wait_for_lite_core():
            """Waits for the lite Hive core and enables basic chat."""
            bootstrap_instance.lite_core_ready.wait() # type: ignore
            return gr.Textbox(placeholder=f"Talk to {CFG['AGENT_NAME']} (Lite Mode)", interactive=True)

        demo.load(wait_for_lite_core, None, [msg])
        vocab_btn.click(do_get_vocab_word, [uid_state], [vocab_output]) # type: ignore
        
        def get_hive_instance():
            global HIVE_INSTANCE

            # If the full hive is ready, ensure we are using it, and it's a valid instance.
            if bootstrap_instance.hive_ready.is_set(): # type: ignore
                if bootstrap_instance.hive_instance is not None and (HIVE_INSTANCE is None or HIVE_INSTANCE.lite_mode):
                    HIVE_INSTANCE = bootstrap_instance.hive_instance
                    print("[UI] Full Hive instance attached.")
                return HIVE_INSTANCE
            # type: ignore
            # Otherwise, use the lite instance.
            if HIVE_INSTANCE is None:
                if bootstrap_instance.lite_core_ready.is_set() and bootstrap_instance.hive_lite_instance is not None:
                    HIVE_INSTANCE = bootstrap_instance.hive_lite_instance
                    print("[UI] Using Lite Hive instance while full core initializes.")
                else:
                    # Neither lite nor full is ready.
                    return None
            return HIVE_INSTANCE



        def wait_for_voice_features(request: gr.Request):
            """Waits for ASR/TTS models and enables voice-related UI elements."""
            bootstrap_instance.voice_ready.wait() # type: ignore
            bootstrap_instance.hive_ready.wait() # Also wait for full core for voice features # type: ignore
            hive_instance = get_hive_instance(bootstrap_instance)

            voice_ready = not hive_instance.lite_mode and hasattr(hive_instance, 'asr_service') and hasattr(hive_instance, 'tts_service')
            video_ready = not hive_instance.lite_mode and hasattr(hive_instance, 'video_service') and CFG["VIDEO_ENABLED"] # type: ignore

            return (
                gr.Markdown("✅ Voice models ready.", visible=voice_ready),
                gr.Audio(interactive=voice_ready), # ptt_audio_in
                gr.Textbox(interactive=voice_ready), # ptt_transcript
                gr.Button(interactive=voice_ready), # ptt_transcribe_btn
                gr.Button(interactive=voice_ready), # ptt_chat_btn
                gr.Button(interactive=voice_ready), # vocal_chat_btn
                gr.Audio(interactive=voice_ready), # enroll_audio
                gr.Button(interactive=voice_ready), # enroll_btn
                gr.Button(interactive=voice_ready), # who_btn
                gr.Markdown("✅ Camera ready." if video_ready else "Camera disabled or not found.", visible=True),
                gr.Image(interactive=video_ready), # video_out
            )
        demo.load(wait_for_voice_features, None, [voice_status_md, ptt_audio_in, ptt_transcript, ptt_transcribe_btn, ptt_chat_btn, vocal_chat_btn, enroll_audio, enroll_btn, who_btn, camera_status_md, video_out], show_progress="hidden")
        def stream_video():
            """Streams video frames from the VideoService to the UI."""
            hive_instance = get_hive_instance(bootstrap_instance) # type: ignore
            if not (
                hive_instance and not hive_instance.lite_mode and 
                hasattr(hive_instance, 'video_service') and hive_instance.video_service and
                CFG["VIDEO_ENABLED"]
            ):
                yield None
                return

            video_service = hive_instance.video_service
            while not video_service.stop_event.is_set():
                frame = video_service.get_frame()
                if frame is not None:
                    yield cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                time.sleep(0.05) # ~20 fps
        demo.load(stream_video, None, video_out)

        def do_online_update():
            hive_instance = get_hive_instance(bootstrap_instance) # type: ignore
            if hive_instance.lite_mode: return "Online features are disabled in Lite Mode." # type: ignore
            return "Added %s new summaries to curves." % (hive_instance.online_update().get("added",0))

        connect_now.click(lambda: (NET.kick_async() or "Auto-connect started in background."), [], [wifi_status]) # type: ignore
        online_now.click(do_online_update, [], [online_status])

        def on_login_or_mode_change(role, pick): # type: ignore
            is_adm = is_admin(pick, role)
            return gr.Tab(visible=is_adm)

        # This function is now the core of the hands-free mode, using the new VADService.
        def process_vocal_chat_stream(stream, state, uid, role, mode, chatbot_history, request: gr.Request): # type: ignore
            now = time.time() # type: ignore
            hive_instance = get_hive_instance() # type: ignore
            if hive_instance.lite_mode or not hasattr(hive_instance, 'vad_service') or not hive_instance.vad_service: # type: ignore
                return None, state, chatbot_history, "VAD service not ready."

            if stream is None:
                if state["active"] and now - state.get("last_interaction_time", now) > state["conversation_timeout"]:
                    state["active"] = False
                    return None, state, chatbot_history, "Status: Sleeping. Say wake word to start."
                return None, state, chatbot_history, state.get("status_text", "Status: Inactive")

            if not state["active"]:
                return None, state, chatbot_history, "Status: Sleeping. Say wake word to start."

            sampling_rate, audio_chunk = stream
            
            # Use the VAD service to get speech segments
            for speech_segment in hive_instance.vad_service.process_stream(audio_chunk): # type: ignore
                state["last_interaction_time"] = now
                yield None, state, chatbot_history, "Status: Transcribing..."

                with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmpfile:
                    sf.write(tmpfile.name, speech_segment, sampling_rate)
                    asr_result = hive_instance.asr_service.transcribe(tmpfile.name, uid) # type: ignore
                    os.remove(tmpfile.name)

                user_text = asr_result["text"]
                if not user_text:
                    continue

                chatbot_history = (chatbot_history or []) + [[user_text, "..."]]
                yield None, state, chatbot_history, "Status: Thinking..."

                eff_role = role if mode == "admin" else "user"
                final_message, intent = hive_instance._prepare_chat_input(user_text, "en", False, None) # type: ignore
                max_tokens = 1024 if intent == "essay_review" else 1024 # Increased for longer responses
                full_prompt = hive_instance.compiler.compile(final_message, [], intent=intent) # type: ignore

                full_reply = ""
                sentence_buffer = ""
                for token in hive_instance.chat_stream(full_prompt, max_new_tokens=max_tokens, temperature=0.7): # type: ignore
                    full_reply += token
                    sentence_buffer += token
                    chatbot_history[-1][1] = full_reply.strip()
                    
                    match = re.search(r'([^.!?]+[.!?])', sentence_buffer)
                    if match:
                        sentence_to_speak = match.group(0).strip()
                        sentence_buffer = sentence_buffer[len(sentence_to_speak):].lstrip()
                        reply_audio_path = hive_instance.tts_service.synthesize(sentence_to_speak, uid) # type: ignore
                        yield gr.Audio(value=reply_audio_path, autoplay=True), state, chatbot_history, "Status: Speaking..."
                
                if sentence_buffer.strip():
                    reply_audio_path = hive_instance.tts_service.synthesize(sentence_buffer, uid) # type: ignore
                    yield gr.Audio(value=reply_audio_path, autoplay=True), state, chatbot_history, "Status: Speaking..."

            state["last_interaction_time"] = time.time()
            yield None, state, chatbot_history, "Status: Active, listening for follow-up..."

        def toggle_vocal_chat(state):
            state["active"] = not state["active"]
            status_text = "Status: Active, listening..." if state["active"] else "Status: Inactive"
            btn_text = "Stop Hands-Free Conversation" if state["active"] else "Start Hands-Free Conversation"
            
            # Toggle visibility of the streaming mic
            mic_visibility = state["active"]
            
            return state, status_text, gr.Button(value=btn_text), gr.Audio(visible=mic_visibility, streaming=True)

        vocal_chat_btn.click(toggle_vocal_chat, [vocal_chat_state], [vocal_chat_state, vocal_chat_status, vocal_chat_btn, vocal_mic])

        # --- Wake Word Detection Logic ---
        porcupine_instance = None
        if _HAVE_PVP and CFG.get("PVPORCUPINE_ACCESS_KEY"): # type: ignore
            keyword_paths: List[str] = []
            keywords = [k.strip() for k in CFG["HIVE_WAKE_WORDS"].split(',') if k.strip()] # type: ignore
            
            for keyword in keywords:
                custom_path = os.path.join(CFG["HIVE_HOME"], "keywords", f"{keyword}_{_os_name()}.ppn")
                if os.path.exists(custom_path):
                    keyword_paths.append(custom_path)
                elif keyword in pvporcupine.BUILTIN_KEYWORDS: # type: ignore
                    keyword_paths.append(keyword)
            
            if not keyword_paths: keyword_paths = ['bumblebee']

            try:
                porcupine_instance = pvporcupine.create( # type: ignore
                    access_key=CFG["PVPORCUPINE_ACCESS_KEY"], # type: ignore
                    keyword_paths=keyword_paths
                )
                print(f"[WakeWord] Listening for: {keywords}")
            except Exception as e:
                print(f"[WakeWord] Error initializing Porcupine: {e}. Wake word will be disabled.")
                porcupine_instance = None
        
        # Auto-start wake word listener on Pi
        is_pi = 'raspberrypi' in platform.machine().lower()
        if is_pi and porcupine_instance:
            print("[WakeWord] Raspberry Pi detected. Wake word listener is always on.")

        def process_wake_word_stream(stream, ww_state, vc_state, request: gr.Request): # type: ignore
            if not porcupine_instance or stream is None or vc_state.get("active", False):
                return ww_state, vc_state, "Status: Inactive", gr.Button(value="Start Hands-Free Conversation")

            sampling_rate, audio_chunk = stream
            # Porcupine expects 16-bit integers
            audio_int16 = (audio_chunk * 32767).astype(np.int16)
            ww_state["buffer"] += audio_int16.tobytes()

            frame_length = porcupine_instance.frame_length # type: ignore
            while len(ww_state["buffer"]) >= frame_length * 2: # 2 bytes per int16
                frame_bytes = ww_state["buffer"][:frame_length * 2]
                ww_state["buffer"] = ww_state["buffer"][frame_length * 2:]
                frame = struct.unpack_from("h" * frame_length, frame_bytes)
                
                keyword_index = porcupine_instance.process(frame) # type: ignore
                if keyword_index >= 0:
                    print(f"[WakeWord] Detected wake word! Activating hot mic.")
                    vc_state["active"] = True
                    vc_state["last_interaction_time"] = time.time() # Start conversation timer
                    status_text = "Status: Wake word detected! Listening for command..."
                    return ww_state, vc_state, status_text, gr.Button(value="Stop Vocal Chat")
            return ww_state, vc_state, "Status: Inactive", gr.Button(value="Start Hands-Free Conversation")

        if porcupine_instance:
            wake_word_mic.stream(process_wake_word_stream, [wake_word_mic, wake_word_state, vocal_chat_state], [wake_word_state, vocal_chat_state, vocal_chat_status, vocal_chat_btn])

        def is_admin(mode, role): return (mode == "admin") and (role in ("admin_general", "admin_super", "owner"))

        def do_add(mode, role, caller, nm, rl, pw): # type: ignore
            if not is_admin(mode, role): return "Switch to Admin mode to use this."
            d=_load_users(); cu,_=_find_user(d, caller or "")
            if not cu: return "Login first as admin."
            if rl not in PERMS.get(cu["role"],{}).get("can_add",[]): return f"{cu['role']} cannot add {rl}."
            uid=f"{rl}:{int(time.time())}"
            entry={"id":uid,"name":nm,"role":rl,"pass":pw if rl!='user' else "", "prefs":{"activation_names":[CFG["AGENT_NAME"]],"language":"en"}} # type: ignore
            if rl=="owner":
                for group in ["admins_super", "admins_general", "users"]:
                    d[group] = [u for u in d.get(group, []) if u.get("id") != d.get("owner", {}).get("id")]
                d["owner"] = entry
            elif rl=="admin_super": d["admins_super"].append(entry)
            elif rl=="admin_general": d["admins_general"].append(entry)
            else: d["users"].append(entry)
            _save_json(USERS_DB,d); return f"Added {rl}: {nm}"
        add_btn.click(do_add, [mode_state, role_state, uid_state, add_name, add_role, add_pass], [out_add])

        def do_rename(mode, role, caller, tgt, nm): # type: ignore
            if not is_admin(mode, role): return "Switch to Admin mode to use this."
            d=_load_users(); u,_=_find_user(d, tgt or "")
            if not u: return "Target not found."
            cu,_=_find_user(d, caller or "")
            if not cu: return "Login first."
            if u.get("role") in PERMS.get(cu.get("role"),{}).get("can_edit_profile_of",[]):
                u["name"]=nm; _save_json(USERS_DB,d); return "Renamed."
            return "Not allowed."
        rename_btn.click(do_rename,[mode_state, role_state, uid_state, target, new_name],[out])

        def do_pass(mode, role, caller, tgt, pw): # type: ignore
            if not is_admin(mode, role): return "Switch to Admin mode to use this."
            d=_load_users(); u,_=_find_user(d, tgt or "")
            if not u: return "Target not found."
            cu,_=_find_user(d, caller or "")
            if not cu: return "Login first."
            if u.get("role") in PERMS.get(cu.get("role"),{}).get("can_edit_profile_of",[]):
                u["pass"]=pw; _save_json(USERS_DB,d); return "Password changed."
            return "Not allowed."
        pass_btn.click(do_pass,[mode_state, role_state, uid_state, target, new_pass],[out])

        def do_role(mode, role, caller, tgt, rl): # type: ignore
            if not is_admin(mode, role): return "Switch to Admin mode to use this."
            d=_load_users(); u,_=_find_user(d, tgt or "")
            if not u: return "Target not found."
            cu,_=_find_user(d, caller or ""); 
            if not cu: return "Login first."
            allowed_new = {"owner":["owner","admin_super","admin_general","user"],
                           "admin_super":["admin_super","admin_general","user"],
                           "admin_general":["admin_general","user"]}.get(cu.get("role"), [])
            if u.get("role") not in PERMS.get(cu.get("role"),{}).get("can_edit_role_of",[]) or rl not in allowed_new:
                return f"Not allowed to set {rl}."
            for grp in ["admins_super","admins_general","users"]:
                if grp in d:
                    d[grp] = [user for user in d[grp] if user.get("id") != u.get("id")]
            if rl=="owner": d["owner"]=u; u["role"]="owner"
            elif rl=="admin_super": d["admins_super"].append(u); u["role"]="admin_super"
            elif rl=="admin_general": d["admins_general"].append(u); u["role"]="admin_general"
            else: d["users"].append(u); u["role"]="user"
            _save_json(USERS_DB,d); return f"Role set to {rl}."
        role_btn.click(do_role,[mode_state, role_state, uid_state, target, new_role],[out])

        def run_ingest_background(hive_instance): # type: ignore
            """
            Triggers the background ingestion process.
            """
            if hive_instance.lite_mode: return "Ingestion is disabled in Lite Mode."
            def ingest_task(): # type: ignore
                staged_ingest_chain_if_enabled(str(hive_instance.config["CURVE_DIR"]))
            threading.Thread(target=ingest_task, daemon=True).start()
            return "Background ingestion process started. See logs for details."
        ingest_now_btn.click(lambda: run_ingest_background(get_hive_instance()), [], [ingest_status])

        # This function has a potential issue if get_hive_instance() returns a lite instance.
        # It is now guarded with a check.
        def compress_memory(h): # type: ignore
            if h.lite_mode or not hasattr(h, 'store'):
                return "Memory compression is not available until the Full Hive Core is ready."
            ok,msg= _archive_memory(str(h.store.dir))
            return msg
        mem_compress_btn.click(lambda: compress_memory(get_hive_instance()), [], [compress_status])

        def do_hotpatch(mode, role, patch_json): # type: ignore
            """
            Applies a runtime hotpatch from the admin console.
            """
            if not is_admin(mode, role):
                return "Hotpatching is an admin-only feature."
            try: patch=json.loads(patch_json)
            except Exception as e: return f"Invalid JSON: {e}"
            
            hive_instance = get_hive_instance()
            if hive_instance.lite_mode or not hasattr(hive_instance, 'overlay'):
                return "Hotpatching is not available in Lite Mode."
            ok, msg = hive_instance.overlay.patch(patch, actor_role=role)
            return msg
        hotpatch_apply.click(do_hotpatch,[mode_state, role_state, hotpatch_patch],[hotpatch_status])

        # This state will hold the session hash for guest users.
        session_id_state = gr.State(None)
        _last: Dict[str, any] = {"id": None, "obj": None}

        # This function is safe because it's only called by the user on the full UI.
        # It is now guarded with a check.
        def do_apply(role, mode): # type: ignore
            hive_instance = get_hive_instance() # type: ignore
            if hive_instance.lite_mode or not hasattr(hive_instance, 'changes'): return "Change management is disabled in Lite Mode."
            if role not in ("admin_super","owner") or mode!="admin": return "Only admin_super or owner may apply."
            if not _last["obj"]: return "No proposal loaded." # type: ignore
            res=hive_instance.changes.test_and_compare(str(_last["id"]), _last["obj"]) # type: ignore
            if not res.get("ok"): return f"Test failed: {res.get('reason','unknown')}"
            if _last["obj"].kind=="code" and role!="owner" and not CFG["OPT_AUTO_APPLY"]: return "Awaiting Owner approval for code changes." # type: ignore
            ok,msg=hive_instance.changes.apply(res); return msg if ok else f"Apply failed: {msg}" # type: ignore

        def do_propose(kind,name,ver,reason,patch): # type: ignore
            hive_instance = get_hive_instance() # type: ignore
            if hive_instance.lite_mode or not hasattr(hive_instance, 'changes'): return {"status": "Error", "reason": "Proposals disabled in Lite Mode."}
            cp=ChangeProposal(kind=kind,name=name or "",version=ver or "",reason=reason or "",patch_text=patch or "")
            pid=hive_instance.changes.propose(cp); _last["id"]=pid; _last["obj"]=cp # type: ignore
            return {"status": "Proposed", "kind": kind, "name": name or '(code patch)', "id": pid} # type: ignore

        def do_test(): # type: ignore
            if not _last["obj"]: return "No proposal in memory. Submit one first." # type: ignore
            if get_hive_instance().lite_mode or not hasattr(get_hive_instance(), 'changes'): return {"status": "Error", "reason": "Testing disabled in Lite Mode."}
            res=get_hive_instance().changes.test_and_compare(str(_last["id"]), _last["obj"]); return res # type: ignore
        propose_btn.click(do_propose, [prop_kind,prop_name,prop_ver,prop_reason,prop_patch],[opt_out]) # type: ignore
        test_btn.click(lambda: do_test(), [], [opt_out])
        apply_btn.click(do_apply, [role_state, mode_state], [opt_out])
        
    demo.launch(
        server_name="0.0.0.0", 
        server_port=int(os.environ.get("PORT")) if os.environ.get("PORT") else None, 
        share=os.getenv("GRADIO_SHARE", "false").lower() == "true"
    ); return demo

def get_hive_instance(bootstrap_instance: "Bootstrap", lite: Optional[bool] = None, caps: Optional[Dict] = None):
    """
    Global function to safely get the current Hive instance.
    It prioritizes the full instance if ready, otherwise falls back to the lite one.
    """
    global HIVE_INSTANCE
    if bootstrap_instance.hive_ready.is_set() and bootstrap_instance.hive_instance:
        if HIVE_INSTANCE is None or HIVE_INSTANCE.lite_mode:
            HIVE_INSTANCE = bootstrap_instance.hive_instance
            print("[get_hive_instance] Switched to Full Hive Instance.")
    elif HIVE_INSTANCE is None and bootstrap_instance.lite_core_ready.is_set() and bootstrap_instance.hive_lite_instance:
        HIVE_INSTANCE = bootstrap_instance.hive_lite_instance
        print("[get_hive_instance] Using Lite Hive instance.")
    
    if HIVE_INSTANCE is None:
        print("[ERROR] get_hive_instance: No Hive instance is available.")
    return HIVE_INSTANCE


class Bootstrap:
    """Handles the entire application startup sequence cleanly."""
    def __init__(self, config: Dict):
        self.config = config
        self.caps: Optional[Dict] = None
        self.env_detector = EnvDetector()
        self.hive_instance: Optional[Hive] = None
        self.hive_lite_instance: Optional[Hive] = None
        self.hive_ready = threading.Event()
        self.lite_core_ready = threading.Event()
        self.voice_ready = threading.Event()
        self.lite_core_success = True
        self.lite_core_error_msg = ""
        Hive.bootstrap_instance = self
        self.env: Optional[Dict] = None
        self.app: Optional[gr.Blocks] = None # type: ignore
        self.init_status: Dict[str, str] = {}
        self.ui_thread: Optional[threading.Thread] = None

    def initialize_persistent_storage(self, base_path: str):
        """Creates the canonical directory structure as per spec."""
        logging.info(f"Ensuring storage layout at {base_path}...")
        root = _Path(base_path)
        for d in DIRS_TO_CREATE: (root / d).mkdir(parents=True, exist_ok=True)
        """Creates the canonical directory structure as per spec.""" # type: ignore
        logging.info(f"Ensuring storage layout at {base_path}...")
        root = _Path(base_path)
        for d in DIRS_TO_CREATE: (root / d).mkdir(parents=True, exist_ok=True)
        # Create default config if not exists
        if not (root / "system" / "config.json").exists():
            _save_json(root / "system" / "config.json", {"note": "Default config created by Bootstrap."})

    def _run_task(self, name: str, target_func, *args):
        """Wrapper to run an initialization task, logging its status."""
        print(f"[Bootstrap] Starting task: {name}...")
        start_time = time.time()
        self.init_status[name] = "running"
        try:
            target_func(*args)
            duration = time.time() - start_time
            self.init_status[name] = "success"
            print(f"[Bootstrap] Task '{name}' completed successfully in {duration:.2f}s.")
        except Exception as e:
            duration = time.time() - start_time
            self.init_status[name] = f"failed: {e}"
            print(f"[ERROR] Task '{name}' failed after {duration:.2f}s: {e}")

    def run(self):
        """Executes the full startup sequence."""
        print("[Bootstrap] Starting Hive System...")
        self.caps = self.env_detector.probe()
        print(f"[Bootstrap] System capabilities: {self.caps}")
        self.initialize_persistent_storage(self.config["HIVE_HOME"])

        # Enforce resource limits based on environment
        if self.caps.get("is_low_memory"):
            print("[Bootstrap] Low memory detected, enabling ultra-constrained mode.")
            self.config["CTX_TOKENS"] = min(self.config.get("CTX_TOKENS", 2048), 1024)

        self._run_task("lite_core_init", self._init_lite_core)

        # Launch the UI in a background thread so it's not blocking
        self.ui_thread = threading.Thread(target=self.launch, daemon=True)
        self.ui_thread.start()

        # Start full initialization in another background thread
        full_init_thread = threading.Thread(target=self.full_initialization_thread, daemon=True)
        full_init_thread.start()

        # Keep the main thread alive to handle signals and wait for shutdown
        import signal
        signal.signal(signal.SIGINT, self.graceful_shutdown)
        signal.signal(signal.SIGTERM, self.graceful_shutdown)
        
        logging.info("Main thread waiting for termination signal.")
        full_init_thread.join()  # Optionally wait for init to complete
        self.ui_thread.join() # Or wait for UI thread to finish

    def full_initialization_thread(self):
        """Handles all non-blocking, full-feature initializations."""
        print("[Bootstrap] Starting full initialization in background...")

        # Start loading heavy models in parallel
        asr_thread = threading.Thread(target=self._run_task, args=("asr_model_load", get_asr))
        tts_thread = threading.Thread(target=self._run_task, args=("tts_model_load", lambda: get_tts(CFG["TTS_LANG"])))

        asr_thread.start()
        tts_thread.start()

        # --- Other Background Tasks ---
        self._run_task("memory_setup", self.setup_memory)

        # Wait for voice models
        asr_thread.join()
        tts_thread.join()
        self.voice_ready.set()
        logging.info("Voice services ready.")

        # Now, initialize the full Hive instance, which includes the main LLM
        self._run_task("full_core_init", self._init_full_core)
        self.hive_ready.set()
        logging.info("Full Hive Core is ready.")



    def _init_lite_core(self):
        """Initializes the fast, responsive lite core."""
        print("[Bootstrap] Initializing Lite Hive Core...")
        try:
            # This now correctly creates the initial lite instance via the global function
            self.hive_lite_instance = Hive(caps=self.caps, lite=True) # type: ignore
            self.lite_core_success = True
            self.lite_core_error_msg = ""
            self.lite_core_ready.set()
            print("[Bootstrap] Lite Hive Core initialized successfully.")
        except Exception as e:
            self.lite_core_success = False
            self.lite_core_error_msg = f"Failed to initialize Lite Hive Core: {e}"
            print(f"[ERROR] {self.lite_core_error_msg}")
            import traceback
            traceback.print_exc()
            # In case of failure, we still set the event to not hang the UI.
            self.lite_core_ready.set()

    def _init_full_core(self):
        """Initializes all features of the full Hive core."""
        logging.info("Initializing Full Hive Core...") # Added logging
        # This is now correctly calling the global get_hive_instance
        llm_thread = threading.Thread(target=lambda: get_hive_instance(lite=False, caps=self.caps), daemon=True)
        asr_thread = threading.Thread(target=get_asr, daemon=True)
        tts_thread = threading.Thread(target=lambda: get_tts(CFG["TTS_LANG"]), daemon=True)

        llm_thread.start()
        asr_thread.start()
        tts_thread.start()

        # --- Other Background Tasks ---
        self._run_task("memory_setup", self.setup_memory)
        
        # Wait for voice models
        asr_thread.join()
        tts_thread.join()
        self.voice_ready.set()
        logging.info("Voice services ready.")

        # Wait for the main LLM and finalize full core
        llm_thread.join()
        self.hive_instance = get_hive_instance(lite=False) # Ensure full instance is assigned
        self.hive_ready.set() # Set *after* self.hive_instance is correctly assigned
        logging.info("Full Hive Core is ready.")

    def soft_restart(self):
        """Performs a hot-reload of the application logic without restarting the process."""
        logging.info("Performing soft restart (hot-reload)...")
        self.hive_ready.clear()
        self.lite_core_ready.clear()
        self.voice_ready.clear()
        if self.hive_instance:
            self.hive_instance.module_manager.stop_all()
        if self.app and hasattr(self.app, 'close'): # type: ignore
            self.app.close()
            self.ui_thread.join(timeout=5.0)

        import app
        importlib.reload(app)

        logging.info("Re-initializing after hot-reload...")
        self.run()

    def setup_memory(self):
        """Handles memory restoration and staged ingestion."""
        def _memory_task():
            print("[Bootstrap] Starting background memory setup...")
            try:
                ok_restored, restore_msg = restore_curves_if_missing(str(self.config["CURVE_DIR"])) # type: ignore
                with open(os.path.join(self.config["STATE_DIR"], "restore_status.log"), "a", encoding="utf-8") as f:
                    f.write(json.dumps({"ok":bool(ok_restored),"msg":restore_msg,"ts":time.time()})+"\n")
                if ok_restored:
                    logging.info(f"Memory restore status: {restore_msg}")
                else:
                        logging.info("No memory restored, proceeding to staged ingestion in background...")
                        staged_ingest_chain_if_enabled(str(self.config["CURVE_DIR"])) # type: ignore
            except Exception as e:
                    with open(os.path.join(self.config["STATE_DIR"], "restore_error.log"), "a", encoding="utf-8") as f:
                        f.write(f"restore/ingest: {e}\n") # type: ignore
        threading.Thread(target=_memory_task, daemon=True).start()

    def launch(self):
        """Launches the appropriate interface (UI or CLI)."""
        if self.config["LAUNCH_UI"]:
            logging.info("Launching Web UI...")
            self.app = launch_ui(self)
            
            # Add the /health endpoint to the FastAPI app
            if self.app and hasattr(self.app, 'app'):
                @self.app.app.get("/health")
                def health_check():
                    status_report = {}
                    now = time.time()
                    for name, data in self.init_status.items():
                        if data.get("status") == "running":
                            elapsed = now - data.get("start_time", now)
                            remaining = max(0, data.get("estimated_duration", 0) - elapsed)
                            status_report[name] = f"running for {elapsed:.1f}s, est. remaining: {remaining:.1f}s"
                        else:
                            status_report[name] = data.get("status")
                    return status_report

        else: # type: ignore
            logging.info("Launching CLI...")
            self.run_cli_loop()

    def run_cli_loop(self): # type: ignore
        """Runs a command-line interface loop for Hive."""
        self.lite_core_ready.wait()
        print("Hive Lite is ready. Type a message and press Enter (Ctrl+C to exit).")
        print("Full core is initializing in the background...")
        try:
            self.hive_instance = self.hive_lite_instance
            while True:
                s = input("> ").strip()
                if not s: continue
                reply = self.hive_instance.chat(s, effective_role="user", caller_id="cli") # type: ignore
                print(reply)
        except (KeyboardInterrupt, EOFError):
            print("\nExiting Hive CLI.")
            pass

    def graceful_shutdown(self, signum=None, frame=None):
        """Handles SIGINT/SIGTERM for clean shutdown."""
        logging.info("\nGraceful shutdown requested...")
        if self.hive_instance and hasattr(self.hive_instance, 'module_manager'):
            logging.info("Stopping all modules...")
            self.hive_instance.module_manager.stop_all() # type: ignore
            if hasattr(self.hive_instance, 'embedding_worker'):
                self.hive_instance.embedding_worker.stop_event.set() # type: ignore
        if self.video_service:
            self.video_service.stop_event.set()
        gr.close_all()
        logging.info("Exiting.")
        sys.exit(0)

if __name__ == "__main__":
    CFG["LAUNCH_UI"] = True
    os.environ["HIVE_USE_HF_INFERENCE"] = "1"
    bootstrap = Bootstrap(CFG)
    bootstrap.run()