Upload 2 files

app.py

@@ -1,618 +1,319 @@
 """
-yui_worker_app.py — YUI SATELLITE WORKER v2.0
-==============================================
-Dùng cho 4 HF Space phụ.
-
-KHÔNG CẦN NHẬP KEY TAY:
-  Worker tự gọi /api/get-keys từ space chính khi khởi động.
-  Chỉ cần set 2 env vars:
-    MAIN_SPACE_URL  = https://bachdathan-ural-ai.hf.space
-    WORKER_SECRET   = [cùng giá trị với space chính]
-    WORKER_ROLE     = code | strong | creative | backup
-    SELF_URL        = https://[username]-[space-name].hf.space (để self-ping)
-    WORKER_URLS     = url1,url2,url3,url4 (để cross-ping, tùy chọn)
-
-TÓM TẮT TÍNH NĂNG:
-  - Chạy code qua Piston (no key) + Judge0 + local sandbox
-  - AI chain theo role (code/strong/creative/backup)
-  - Stealth ping khác interval theo role
-  - Cross-ping workers khác để giữ cluster alive
-  - UptimeRobot ping /health endpoint
-"""
+URAL AI — Worker Space App
+Nhẹ, nhanh, chỉ làm đúng 1 việc: nhận task từ main space và xử lý AI.
 
-import os, time, random, threading, asyncio
-import httpx
-from fastapi import FastAPI, Request, HTTPException
+ENV cần set trong HF Secrets của mỗi worker:
+  WORKER_SECRET   = (giống main space)
+  WORKER_ROLE     = strong | code | creative | backup
+  MAIN_SPACE_URL  = https://bachdathan-ural-ai.hf.space  (URL main space)
+"""
+import os, time, random, asyncio
+import requests as _req
+from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
-import uvicorn
-
-# ── ENV ───────────────────────────────────────────────────────────
-MAIN_SPACE_URL  = os.getenv("MAIN_SPACE_URL", "").rstrip("/")
-WORKER_SECRET   = os.getenv("WORKER_SECRET", "")
-WORKER_ROLE     = os.getenv("WORKER_ROLE", "backup")
-SELF_URL        = os.getenv("SELF_URL", "").rstrip("/")
-WORKER_URLS     = [u.strip() for u in os.getenv("WORKER_URLS","").split(",") if u.strip()]
-
-# ── AUTO-FETCH KEYS ───────────────────────────────────────────────
-# Worker tự lấy key từ space chính, không cần nhập tay
-_KEYS: dict = {
-    "gemini": [], "openrouter": [], "cerebras": [],
-    "mistral": [], "cohere": [], "tavily": [],
-}
-_keys_loaded = threading.Event()
-_keys_lock   = threading.Lock()
+
+app = FastAPI(title="Yui Worker")
+
+# ── Config ────────────────────────────────────────────────────────
+WORKER_SECRET  = os.environ.get("WORKER_SECRET", "")
+WORKER_ROLE    = os.environ.get("WORKER_ROLE", "backup")
+MAIN_SPACE_URL = os.environ.get("MAIN_SPACE_URL",
+                                "https://bachdathan-ural-ai.hf.space")
+
+# ── Key cache ─────────────────────────────────────────────────────
+_keys: dict = {"gemini": [], "cerebras": [], "openrouter": [],
+               "cohere": [], "mistral": []}
+_keys_at: float = 0.0
+_KEY_TTL = 1500  # 25 phút
+
 
 def _fetch_keys_sync():
-    """Gọi /api/get-keys từ main space để lấy keys."""
-    global _KEYS
-    if not MAIN_SPACE_URL or not WORKER_SECRET:
-        print(f"[WORKER-{WORKER_ROLE}] ⚠️ MAIN_SPACE_URL hoặc WORKER_SECRET chưa set!")
-        _keys_loaded.set()
-        return
-
-    for attempt in range(5):
+    """Lấy API keys từ main space (sync, dùng khi startup)."""
+    global _keys, _keys_at
+    try:
+        r = _req.post(
+            f"{MAIN_SPACE_URL}/api/get-keys",
+            json={"secret": WORKER_SECRET, "role": WORKER_ROLE},
+            timeout=20
+        )
+        if r.status_code == 200:
+            _keys = r.json()
+            _keys_at = time.time()
+            g = len(_keys.get("gemini", []))
+            cb = len(_keys.get("cerebras", []))
+            co = len(_keys.get("cohere", []))
+            print(f"[WORKER-{WORKER_ROLE.upper()}] ✅ Keys: G×{g} CB×{cb} CO×{co}")
+        else:
+            print(f"[WORKER] ⚠️ get-keys trả {r.status_code}")
+    except Exception as e:
+        print(f"[WORKER] ⚠️ get-keys lỗi: {e}")
+
+
+def _get_keys() -> dict:
+    if time.time() - _keys_at > _KEY_TTL:
+        _fetch_keys_sync()
+    return _keys
+
+
+# ── Helpers gọi AI ────────────────────────────────────────────────
+def _build_msgs(system: str, history: list, message: str) -> list:
+    msgs = []
+    if system:
+        msgs.append({"role": "system", "content": system[:1500]})
+    for turn in history[-6:]:
+        if isinstance(turn, (list, tuple)) and len(turn) >= 2:
+            msgs.append({"role": "user",      "content": str(turn[0])[:800]})
+            msgs.append({"role": "assistant", "content": str(turn[1])[:800]})
+    msgs.append({"role": "user", "content": message})
+    return msgs
+
+
+def _call_gemini(keys: list, msgs: list, system: str, max_tok: int = 1000) -> str:
+    for key in random.sample(keys, min(3, len(keys))):
         try:
-            import requests as _rq
-            r = _rq.post(
-                f"{MAIN_SPACE_URL}/api/get-keys",
-                json={"secret": WORKER_SECRET, "role": WORKER_ROLE},
-                timeout=15
+            contents = [{"role": "user" if m["role"] == "user" else "model",
+                         "parts": [{"text": m["content"]}]}
+                        for m in msgs if m["role"] != "system"]
+            if not contents:
+                continue
+            payload: dict = {
+                "contents": contents,
+                "generationConfig": {"maxOutputTokens": max_tok, "temperature": 0.7}
+            }
+            if system:
+                payload["system_instruction"] = {"parts": [{"text": system[:1000]}]}
+            r = _req.post(
+                f"https://generativelanguage.googleapis.com/v1beta/models/"
+                f"gemini-2.5-flash:generateContent?key={key}",
+                json=payload, timeout=30
             )
             if r.status_code == 200:
-                data = r.json()
-                with _keys_lock:
-                    for k in _KEYS:
-                        v = data.get(k, [])
-                        if isinstance(v, list) and v:
-                            _KEYS[k] = v
-                print(f"[WORKER-{WORKER_ROLE}] ✅ Keys loaded: "
-                      f"G×{len(_KEYS['gemini'])} OR×{len(_KEYS['openrouter'])} "
-                      f"CB×{len(_KEYS['cerebras'])} M×{len(_KEYS['mistral'])}")
-                _keys_loaded.set()
-                return
-            else:
-                print(f"[WORKER-{WORKER_ROLE}] ⚠️ get-keys HTTP {r.status_code} — retry {attempt+1}/5")
+                cands = r.json().get("candidates", [])
+                if cands:
+                    text = (cands[0].get("content", {})
+                            .get("parts", [{}])[0].get("text", ""))
+                    if text and len(text) > 5:
+                        return text
         except Exception as e:
-            print(f"[WORKER-{WORKER_ROLE}] ⚠️ get-keys lỗi: {e} — retry {attempt+1}/5")
-        time.sleep(10 * (attempt + 1))
+            print(f"[WORKER] Gemini key fail: {type(e).__name__}")
+    return ""
 
-    print(f"[WORKER-{WORKER_ROLE}] ❌ Không lấy được keys sau 5 lần thử")
-    _keys_loaded.set()
 
+def _call_cerebras(keys: list, msgs: list, max_tok: int = 800) -> str:
+    try:
+        import openai as _oai
+        model = "gpt-oss-120b"
+        client = _oai.OpenAI(api_key=random.choice(keys),
+                             base_url="https://api.cerebras.ai/v1")
+        r = client.chat.completions.create(
+            model=model, messages=msgs,
+            max_tokens=max_tok, temperature=0.7, timeout=25
+        )
+        return r.choices[0].message.content or ""
+    except Exception as e:
+        print(f"[WORKER] Cerebras fail: {type(e).__name__}: {e}")
+        # Thử model nhẹ hơn
+        try:
+            import openai as _oai2
+            client2 = _oai2.OpenAI(api_key=random.choice(keys),
+                                   base_url="https://api.cerebras.ai/v1")
+            r2 = client2.chat.completions.create(
+                model="zai-glm-4.7", messages=msgs,
+                max_tokens=max_tok, temperature=0.7, timeout=20
+            )
+            return r2.choices[0].message.content or ""
+        except Exception:
+            return ""
 
-def _key_refresh_loop():
-    """Cập nhật keys mỗi 6 giờ (key mới được thêm sẽ tự sync)."""
-    time.sleep(21600)  # 6 giờ sau lần đầu
-    while True:
-        _fetch_keys_sync()
-        time.sleep(21600)
-
-
-# ── HELPERS ───────────────────────────────────────────────────────
-def _rkey(name: str):
-    with _keys_lock:
-        pool = _KEYS.get(name, [])
-    return random.choice(pool) if pool else None
-
-def _quota_ok(name: str) -> bool:
-    return time.time() > _quota_hits.get(name, 0)
-
-_quota_hits: dict = {}
-def _quota_hit(name: str, secs: int = 60):
-    _quota_hits[name] = time.time() + secs
-    print(f"[WORKER-{WORKER_ROLE}] ⏳ {name} cooldown {secs}s")
-
-
-# ── ROLE CONFIG ───────────────────────────────────────────────────
-ROLE_CHAIN = {
-    "code":     [("mistral","codestral-latest"), ("openrouter","deepseek/deepseek-v4-flash:free"),
-                 ("openrouter","qwen/qwen3-coder:free"), ("cerebras","qwen-3-235b-a22b-instruct-2507")],
-    "strong":   [("cerebras","qwen-3-235b-a22b-instruct-2507"), ("cerebras","gpt-oss-120b"),
-                 ("openrouter","deepseek/deepseek-r1-0528:free"), ("gemini","gemini-2.5-flash")],
-    "creative": [("gemini","gemini-2.5-flash"), ("openrouter","meta-llama/llama-3.3-70b-instruct:free"),
-                 ("openrouter","qwen/qwen3-32b:free"), ("cohere","command-r-plus-08-2024")],
-    "backup":   [("gemini","gemini-2.5-flash"), ("cerebras","qwen-3-235b-a22b-instruct-2507"),
-                 ("openrouter","deepseek/deepseek-r1:free"), ("cohere","command-r-plus-08-2024")],
-}
-ROLE_TASKS = {
-    "code":     ["code","debug","math","code_review"],
-    "strong":   ["knowledge","science","analysis","reasoning"],
-    "creative": ["creative","writing","rpg","chat","emotional"],
-    "backup":   ["*"],
-}
-
-
-# ── AI CALLERS ────────────────────────────────────────────────────
-async def _call(provider: str, model: str, message: str, system: str, history: list) -> str | None:
-    key = _rkey(provider)
-    if not key and provider not in ("piston",):
-        return None
-    if not _quota_ok(provider):
-        return None
-
-    msgs = [{"role":"system","content":system}]
-    for h in history[-4:]:
-        if isinstance(h,dict): msgs.append(h)
-    msgs.append({"role":"user","content":message})
 
+def _call_cohere(keys: list, msgs: list, message: str, system: str) -> str:
     try:
-        if provider == "gemini":
-            contents = [{"role":"user","parts":[{"text":message}]}]
-            body = {"system_instruction":{"parts":[{"text":system}]},
-                    "contents":contents,
-                    "generationConfig":{"temperature":0.7,"maxOutputTokens":4096}}
-            async with httpx.AsyncClient(timeout=45) as cl:
-                r = await cl.post(
-                    f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={key}",
-                    json=body)
-            if r.status_code==429: _quota_hit("gemini",60); return None
-            if r.status_code!=200: return None
-            return r.json()["candidates"][0]["content"]["parts"][0]["text"]
-
-        elif provider in ("cerebras","openrouter","mistral","cohere"):
-            import openai as _oai
-            base_urls = {
-                "cerebras":   "https://api.cerebras.ai/v1",
-                "openrouter": "https://openrouter.ai/api/v1",
-                "mistral":    "https://api.mistral.ai/v1",
-                "cohere":     "https://api.cohere.ai/compatibility/v1",
-            }
-            cl = _oai.AsyncOpenAI(api_key=key, base_url=base_urls[provider])
-            resp = await cl.chat.completions.create(
-                model=model, messages=msgs, temperature=0.3, max_tokens=8192)
-            return resp.choices[0].message.content
-
+        import cohere as _co
+        client = _co.Client(api_key=random.choice(keys))
+        hist = []
+        for m in msgs[:-1]:
+            if m["role"] == "system":
+                continue
+            hist.append({"role": "USER" if m["role"] == "user" else "CHATBOT",
+                         "message": m["content"][:400]})
+        r = client.chat(
+            model="command-r-08-2024",
+            message=message,
+            chat_history=hist or None,
+            preamble=system[:500] if system else None,
+            max_tokens=600,
+        )
+        return r.text or ""
     except Exception as e:
-        err = str(e).lower()
-        if "429" in err or "quota" in err or "rate" in err:
-            _quota_hit(provider, 90)
-        print(f"[WORKER-{WORKER_ROLE}] {provider} err: {str(e)[:60]}")
-    return None
-
-
-async def process(message: str, system: str, history: list, task: str) -> str:
-    chain = ROLE_CHAIN.get(WORKER_ROLE, ROLE_CHAIN["backup"])
-    for provider, model in chain:
-        if not _quota_ok(provider): continue
-        result = await _call(provider, model, message, system, history)
-        if result and len(result.strip()) > 5:
-            print(f"[WORKER-{WORKER_ROLE}] ✅ {provider}/{model[:25]}")
-            return result
-    return "⚠️ Worker tạm hết quota — space chính xử lý tiếp."
-
-
-# ── CODE EXECUTION (Piston, không cần key) ───────────────────────
-PISTON_LANGS = {
-    "python":"python","py":"python","javascript":"javascript","js":"javascript",
-    "typescript":"typescript","ts":"typescript","java":"java","c":"c","cpp":"c++",
-    "c++":"c++","rust":"rust","go":"go","ruby":"ruby","php":"php","bash":"bash",
-    "lua":"lua","kotlin":"kotlin","r":"r","csharp":"csharp","cs":"csharp",
-}
-PISTON_VERS = {
-    "python":"3.10.0","javascript":"18.15.0","typescript":"5.0.3","java":"15.0.2",
-    "c":"10.2.0","c++":"10.2.0","rust":"1.50.0","go":"1.16.2","ruby":"3.0.1",
-    "php":"8.2.3","bash":"5.2.0","lua":"5.4.4","kotlin":"1.8.20","r":"4.1.1","csharp":"6.12.0",
-}
-_RATE_DATA: dict = {}
-_RATE_LOCK = threading.Lock()
-
-def _rate_ok(uid: str, is_admin: bool) -> tuple[bool,str]:
-    if is_admin: return True,""
-    now = time.time()
-    with _RATE_LOCK:
-        d = _RATE_DATA.setdefault(uid,{"m":[],"h":[]})
-        d["m"] = [t for t in d["m"] if now-t<60]
-        d["h"] = [t for t in d["h"] if now-t<3600]
-        if len(d["m"]) >= 3:
-            return False, f"⏳ Quá 3 lần/phút — chờ {int(60-(now-d['m'][0]))+1}s!"
-        if len(d["h"]) >= 20:
-            return False, f"⏳ Quá 20 lần/giờ — chờ {int(3600-(now-d['h'][0]))//60+1} phút!"
-        d["m"].append(now); d["h"].append(now)
-    return True,""
-
-async def run_code_piston(code: str, lang: str, stdin: str = "") -> dict:
-    pl = PISTON_LANGS.get(lang.lower())
-    if not pl: return {"ok":False,"error":f"Piston không hỗ trợ: {lang}","output":""}
-    ver = PISTON_VERS.get(pl,"0.0.0")
+        print(f"[WORKER] Cohere fail: {type(e).__name__}")
+        return ""
+
+
+def _call_openrouter(keys: list, msgs: list, task: str = "chat") -> str:
     try:
-        async with httpx.AsyncClient(timeout=20) as cl:
-            r = await cl.post("https://emkc.org/api/v2/piston/execute",
-                json={"language":pl,"version":ver,
-                      "files":[{"name":"main","content":code}],"stdin":stdin})
-        if r.status_code==429: return {"ok":False,"error":"Piston 429 (5/s limit)","output":""}
-        if r.status_code!=200: return {"ok":False,"error":f"Piston HTTP {r.status_code}","output":""}
-        run = r.json().get("run",{})
-        out = (run.get("stdout") or "").strip()
-        err = (run.get("stderr") or "").strip()
-        code_ret = run.get("code",0)
-        if code_ret!=0 and not out:
-            return {"ok":False,"error":err[:500],"output":""}
-        return {"ok":True,"output":out[:1500],"error":err[:100],"backend":"piston"}
+        import openai as _oai
+        model = ("qwen/qwen3-coder:free" if task in ("code", "math", "debug")
+                 else "openrouter/owl-alpha")
+        client = _oai.OpenAI(api_key=random.choice(keys),
+                             base_url="https://openrouter.ai/api/v1")
+        r = client.chat.completions.create(
+            model=model, messages=msgs,
+            max_tokens=600, timeout=20
+        )
+        return r.choices[0].message.content or ""
     except Exception as e:
-        return {"ok":False,"error":str(e)[:80],"output":""}
+        print(f"[WORKER] OpenRouter fail: {type(e).__name__}")
+        return ""
 
 
-# ── FASTAPI APP ───────────────────────────────────────────────────
-from contextlib import asynccontextmanager
+# ── Endpoints ─────────────────────────────────────────────────────
 
-@asynccontextmanager
-async def lifespan(app_instance):
-    # Startup
-    print(f"🚀 Yui Worker [{WORKER_ROLE}] khởi động — URL: {SELF_URL or 'chưa set'}")
-    for fn in [_fetch_keys_sync, _key_refresh_loop, _self_ping, _cross_ping]:
-        threading.Thread(target=fn, daemon=True).start()
-    yield
-    # Shutdown (nothing to cleanup)
+@app.get("/")
+@app.head("/")
+async def root():
+    return {"yui": "worker", "role": WORKER_ROLE, "status": "ready"}
 
-app = FastAPI(title=f"Yui Worker [{WORKER_ROLE}]", docs_url=None, redoc_url=None, lifespan=lifespan)
 
-# ══════════════════════════════════════════════════════════════════
-# 🔍 A. WEB SEARCH ENDPOINT — dùng Jina Reader + DuckDuckGo (free)
-# ══════════════════════════════════════════════════════════════════
-@app.post("/api/search")
-async def search_endpoint(request: Request):
-    """Web search worker — Jina Reader + DuckDuckGo, không cần key."""
-    try: body = await request.json()
-    except Exception: raise HTTPException(400, "Invalid JSON")
-    import hmac as _hm_s
-    if not WORKER_SECRET or not _hm_s.compare_digest(
-        body.get("secret","").encode(), WORKER_SECRET.encode()
-    ): raise HTTPException(403, "Unauthorized")
-
-    query   = body.get("query", "").strip()
-    mode    = body.get("mode", "search")   # "search" | "fetch"
-    url_in  = body.get("url", "")
-    max_r   = min(int(body.get("max_results", 5)), 8)
-    if not query and not url_in:
-        raise HTTPException(400, "query hoặc url required")
+@app.get("/health")
+@app.head("/health")
+async def health():
+    return {"status": "ok", "role": WORKER_ROLE,
+            "has_keys": bool(_keys.get("gemini"))}
 
+
+@app.post("/api/worker")
+async def worker_chat(request: Request):
+    """Endpoint chính — nhận task từ main space và trả kết quả."""
     t0 = time.time()
-    results = []
 
-    if mode == "fetch" and url_in:
-        # Jina Reader: fetch nội dung 1 URL cụ thể
-        try:
-            async with httpx.AsyncClient(timeout=20) as cl:
-                r = await cl.get(
-                    f"https://r.jina.ai/{url_in}",
-                    headers={"Accept": "text/plain", "X-No-Cache": "true"},
-                )
-            results = [{"title": url_in, "content": r.text[:3000], "url": url_in}]
-        except Exception as e:
-            results = [{"error": str(e)[:100]}]
+    # Parse body
+    try:
+        body = await request.json()
+    except Exception:
+        return JSONResponse({"ok": False, "error": "invalid json"}, status_code=400)
 
-    else:
-        # DuckDuckGo search (không cần key)
-        try:
-            from urllib.parse import quote
-            async with httpx.AsyncClient(timeout=15) as cl:
-                r = await cl.get(
-                    "https://api.duckduckgo.com/",
-                    params={"q": query, "format": "json", "no_html": 1,
-                            "skip_disambig": 1, "no_redirect": 1},
-                    headers={"User-Agent": "Mozilla/5.0"},
-                )
-            data = r.json()
-            # AbstractText
-            if data.get("AbstractText"):
-                results.append({"title": data.get("Heading",""),
-                                 "content": data["AbstractText"][:1000],
-                                 "url": data.get("AbstractURL","")})
-            # RelatedTopics
-            for t in data.get("RelatedTopics", [])[:max_r]:
-                if isinstance(t, dict) and t.get("Text"):
-                    results.append({"title": t.get("Text","")[:80],
-                                     "content": t.get("Text","")[:500],
-                                     "url": t.get("FirstURL","")})
-            # Fallback: Jina search nếu DDG trả về rỗng
-            if not results:
-                async with httpx.AsyncClient(timeout=20) as cl2:
-                    r2 = await cl2.get(
-                        f"https://s.jina.ai/{quote(query)}",
-                        headers={"Accept": "application/json",
-                                 "X-Retain-Images": "none"},
-                    )
-                if r2.status_code == 200:
-                    for line in r2.text.split("\n\n")[:max_r]:
-                        if line.strip():
-                            results.append({"content": line.strip()[:800], "url": ""})
-        except Exception as e:
-            results = [{"error": str(e)[:100]}]
+    # Auth
+    if not WORKER_SECRET or body.get("secret") != WORKER_SECRET:
+        return JSONResponse({"ok": False, "error": "unauthorized"}, status_code=403)
 
-    return JSONResponse({
-        "ok": True, "query": query, "results": results[:max_r],
-        "elapsed": round(time.time()-t0, 2), "worker": WORKER_ROLE,
-    })
+    message = body.get("message", "").strip()
+    system  = body.get("system", "")
+    history = body.get("history", [])
+    task    = body.get("task", "chat")
 
+    if not message:
+        return JSONResponse({"ok": False, "error": "empty message"}, status_code=400)
 
-# ══════════════════════════════════════════════════════════════════
-# 🔊 C. TTS ENDPOINT — Edge TTS (Microsoft, hoàn toàn free)
-# ══════════════════════════════════════════════════════════════════
-@app.post("/api/tts")
-async def tts_endpoint(request: Request):
-    """Text-to-speech — Edge TTS, không cần key, hỗ trợ giọng Việt."""
-    try: body = await request.json()
-    except Exception: raise HTTPException(400, "Invalid JSON")
-    import hmac as _hm_t
-    if not WORKER_SECRET or not _hm_t.compare_digest(
-        body.get("secret","").encode(), WORKER_SECRET.encode()
-    ): raise HTTPException(403, "Unauthorized")
-
-    text  = body.get("text", "").strip()[:2000]
-    voice = body.get("voice", "vi-VN-HoaiMyNeural")  # default: giọng nữ Việt
-    if not text:
-        raise HTTPException(400, "text required")
+    # Lấy keys
+    keys = _get_keys()
+    msgs = _build_msgs(system, history, message)
 
-    t0 = time.time()
-    try:
-        import edge_tts, io
-        communicate = edge_tts.Communicate(text, voice)
-        audio_data   = io.BytesIO()
-        async for chunk in communicate.stream():
-            if chunk["type"] == "audio":
-                audio_data.write(chunk["data"])
-        audio_bytes = audio_data.getvalue()
-        import base64
-        audio_b64 = base64.b64encode(audio_bytes).decode()
+    reply = ""
+
+    # Chọn model theo role và task
+    gemini = keys.get("gemini", [])
+    cerebras = keys.get("cerebras", [])
+    cohere = keys.get("cohere", [])
+    openrouter = keys.get("openrouter", [])
+
+    # Gemini Flash — ưu tiên hàng đầu
+    if gemini and not reply:
+        max_tok = 1500 if task in ("code", "math", "debug") else 800
+        reply = _call_gemini(gemini, msgs, system, max_tok)
+
+    # Cerebras — fallback mạnh
+    if cerebras and not reply:
+        max_tok = 1200 if task in ("code", "math", "debug") else 700
+        reply = _call_cerebras(cerebras, msgs, max_tok)
+
+    # Cohere — fallback trung bình
+    if cohere and not reply:
+        reply = _call_cohere(cohere, msgs, message, system)
+
+    # OpenRouter — fallback cuối
+    if openrouter and not reply:
+        reply = _call_openrouter(openrouter, msgs, task)
+
+    elapsed = round(time.time() - t0, 2)
+
+    if reply:
         return JSONResponse({
-            "ok": True, "audio_b64": audio_b64,
-            "format": "mp3", "voice": voice,
-            "chars": len(text),
-            "elapsed": round(time.time()-t0, 2),
+            "ok": True,
+            "reply": reply,
+            "role": WORKER_ROLE,
+            "elapsed": elapsed
         })
-    except ImportError:
-        return JSONResponse({"ok": False, "error": "edge_tts chưa cài. Thêm vào requirements.txt"})
-    except Exception as e:
-        return JSONResponse({"ok": False, "error": str(e)[:200]})
-
-
-# ══════════════════════════════════════════════════════════════════
-# 💾 E. SMART CACHE ENDPOINT — SQLite-based semantic cache
-# ══════════════════════════════════════════════════════════════════
-import sqlite3, hashlib, json as _json_mod
-
-_CACHE_DB   = "/tmp/yui_cache.db"
-_CACHE_LOCK = threading.Lock()
-
-def _cache_init():
-    with sqlite3.connect(_CACHE_DB) as db:
-        db.execute("""CREATE TABLE IF NOT EXISTS cache (
-            key      TEXT PRIMARY KEY,
-            query    TEXT,
-            reply    TEXT,
-            task     TEXT,
-            hits     INTEGER DEFAULT 0,
-            created  REAL,
-            expires  REAL
-        )""")
-        db.execute("CREATE INDEX IF NOT EXISTS idx_expires ON cache(expires)")
-        db.commit()
-
-def _cache_key(query: str, task: str = "*") -> str:
-    # Normalize: lowercase, strip spaces thừa
-    norm = " ".join(query.lower().split())
-    return hashlib.sha256(f"{task}:{norm}".encode()).hexdigest()[:32]
-
-def _cache_get(query: str, task: str = "*") -> "str | None":
-    key = _cache_key(query, task)
-    now = time.time()
-    with _CACHE_LOCK:
-        try:
-            with sqlite3.connect(_CACHE_DB) as db:
-                row = db.execute(
-                    "SELECT reply FROM cache WHERE key=? AND expires>?",
-                    (key, now)
-                ).fetchone()
-                if row:
-                    db.execute("UPDATE cache SET hits=hits+1 WHERE key=?", (key,))
-                    db.commit()
-                    return row[0]
-        except Exception:
-            pass
-    return None
+    else:
+        return JSONResponse({
+            "ok": False,
+            "error": "all models failed",
+            "role": WORKER_ROLE,
+            "elapsed": elapsed
+        }, status_code=503)
 
-def _cache_set(query: str, reply: str, task: str = "*", ttl: int = 3600):
-    key  = _cache_key(query, task)
-    now  = time.time()
-    with _CACHE_LOCK:
-        try:
-            with sqlite3.connect(_CACHE_DB) as db:
-                db.execute("""INSERT OR REPLACE INTO cache
-                    (key, query, reply, task, hits, created, expires)
-                    VALUES (?,?,?,?,0,?,?)""",
-                    (key, query[:300], reply, task, now, now+ttl))
-                # Dọn cache hết hạn (max 500 entries)
-                db.execute("DELETE FROM cache WHERE expires<?", (now,))
-                db.execute("""DELETE FROM cache WHERE key NOT IN (
-                    SELECT key FROM cache ORDER BY expires DESC LIMIT 500)""")
-                db.commit()
-        except Exception:
-            pass
-
-# Init cache DB khi startup
-_cache_init()
-
-@app.post("/api/cache")
-async def cache_endpoint(request: Request):
-    """SmartCache — GET hoặc SET cache response."""
-    try: body = await request.json()
-    except Exception: raise HTTPException(400, "Invalid JSON")
-    import hmac as _hm_c
-    if not WORKER_SECRET or not _hm_c.compare_digest(
-        body.get("secret","").encode(), WORKER_SECRET.encode()
-    ): raise HTTPException(403, "Unauthorized")
-
-    action = body.get("action", "get")   # "get" | "set" | "stats"
-    query  = body.get("query", "")
-    task   = body.get("task", "*")
-
-    if action == "get":
-        cached = _cache_get(query, task)
-        return JSONResponse({"ok": True, "hit": cached is not None,
-                             "reply": cached, "worker": WORKER_ROLE})
-
-    elif action == "set":
-        reply = body.get("reply", "")
-        ttl   = int(body.get("ttl", 3600))
-        if query and reply:
-            _cache_set(query, reply, task, ttl)
-        return JSONResponse({"ok": True, "stored": bool(query and reply)})
-
-    elif action == "stats":
-        with _CACHE_LOCK:
-            try:
-                with sqlite3.connect(_CACHE_DB) as db:
-                    count = db.execute("SELECT COUNT(*) FROM cache WHERE expires>?",
-                                       (time.time(),)).fetchone()[0]
-                    hits  = db.execute("SELECT SUM(hits) FROM cache").fetchone()[0] or 0
-            except Exception:
-                count, hits = 0, 0
-        return JSONResponse({"ok": True, "entries": count, "total_hits": hits,
-                             "worker": WORKER_ROLE})
-
-    return JSONResponse({"ok": False, "error": "action phải là get/set/stats"})
-
-
-# ══════════════════════════════════════════════════════════════════
-# 🧠 F. BACKUP BRAIN ENDPOINT — Mini pipeline khi space chính bận
-# ══════════════════════════════════════════════════════════════════
-@app.post("/api/backup-brain")
-async def backup_brain_endpoint(request: Request):
-    """
-    Backup Brain — xử lý request khi space chính timeout/bận.
-    Pipeline đơn giản: không FAISS, không IMAD, không image gen.
-    Chỉ cần trả lời nhanh và đúng.
-    """
-    try: body = await request.json()
-    except Exception: raise HTTPException(400, "Invalid JSON")
-    import hmac as _hm_b
-    if not WORKER_SECRET or not _hm_b.compare_digest(
-        body.get("secret","").encode(), WORKER_SECRET.encode()
-    ): raise HTTPException(403, "Unauthorized")
 
-    message = body.get("message", "").strip()
-    history = body.get("history", [])
-    task    = body.get("task", "*")
-    if not message:
-        raise HTTPException(400, "message required")
+@app.post("/api/tts")
+async def worker_tts(request: Request):
+    """TTS endpoint — edge-tts."""
+    try:
+        body = await request.json()
+        if body.get("secret") != WORKER_SECRET:
+            return JSONResponse({"ok": False}, status_code=403)
+        text = body.get("text", "").strip()
+        if not text:
+            return JSONResponse({"ok": False, "error": "no text"})
+        import edge_tts, base64
+        voice = "vi-VN-HoaiMyNeural"
+        comm = edge_tts.Communicate(text[:600], voice)
+        chunks = []
+        async for chunk in comm.stream():
+            if chunk["type"] == "audio":
+                chunks.append(chunk["data"])
+        audio_b64 = base64.b64encode(b"".join(chunks)).decode()
+        return JSONResponse({"ok": True, "audio_b64": audio_b64})
+    except Exception as e:
+        return JSONResponse({"ok": False, "error": str(e)})
 
-    t0 = time.time()
 
-    # 1. Check SmartCache trước (nếu có cache → trả ngay)
-    cached = _cache_get(message, task)
-    if cached:
-        return JSONResponse({"ok": True, "reply": cached, "source": "cache",
-                             "elapsed": round(time.time()-t0, 2)})
+@app.post("/api/search")
+async def worker_search(request: Request):
+    """Search endpoint — Tavily."""
+    try:
+        body = await request.json()
+        if body.get("secret") != WORKER_SECRET:
+            return JSONResponse({"ok": False}, status_code=403)
+        query = body.get("query", "")
+        tavily = os.environ.get("TAVILY_KEY", "")
+        if not tavily or not query:
+            return JSONResponse({"ok": False, "results": []})
+        r = _req.post("https://api.tavily.com/search",
+                      json={"api_key": tavily, "query": query, "max_results": 3},
+                      timeout=10)
+        return JSONResponse({"ok": True, "results": r.json().get("results", [])})
+    except Exception as e:
+        return JSONResponse({"ok": False, "results": [], "error": str(e)})
 
-    # 2. Gọi AI với pipeline đơn giản
-    system = (
-        "Bạn là Yui, AI assistant thông minh của Boss. "
-        "Trả lời ngắn gọn, chính xác, bằng tiếng Việt. "
-        "Đây là backup mode — không có tool nào khác."
-    )
-    reply = await process(message, system, history, task)
 
-    # 3. Cache kết quả (TTL 30 phút cho backup brain)
-    if reply and len(reply) > 20:
-        _cache_set(message, reply, task, ttl=1800)
+@app.post("/api/backup-brain")
+async def worker_backup_brain(request: Request):
+    """Backup brain — forward to Cohere/Cerebras khi main space quá tải."""
+    return await worker_chat(request)
 
-    return JSONResponse({
-        "ok": True, "reply": reply, "source": "backup_brain",
-        "role": WORKER_ROLE, "elapsed": round(time.time()-t0, 2),
-    })
 
+# ── Startup ───────────────────────────────────────────────────────
 
-@app.get("/")
-@app.get("/health")
-@app.head("/")
-@app.head("/health")
-async def health():
-    return {
-        "status": "ok", "role": WORKER_ROLE,
-        "keys_loaded": _keys_loaded.is_set(),
-        "keys": {k: len(v) for k,v in _KEYS.items() if v},
-        "quota": {k:"cd" if not _quota_ok(k) else "ok" for k in ["gemini","cerebras","openrouter","mistral"]},
-        "ts": int(time.time()),
-    }
+@app.on_event("startup")
+async def startup():
+    import threading
+    print(f"[WORKER-{WORKER_ROLE.upper()}] 🚀 Khởi động...")
+    # Fetch keys ngay lúc startup (sync trong thread để không block)
+    threading.Thread(target=_fetch_keys_sync, daemon=True).start()
 
-@app.post("/api/worker")
-async def worker_endpoint(request: Request):
-    """Nhận request từ space chính."""
-    try: body = await request.json()
-    except Exception: raise HTTPException(400,"Invalid JSON")
-    import hmac as _hm
-    if not WORKER_SECRET or not _hm.compare_digest(
-        body.get("secret","").encode(), WORKER_SECRET.encode()
-    ): raise HTTPException(403,"Unauthorized")
-    message = body.get("message","")
-    system  = body.get("system","Bạn là Yui, AI assistant thông minh.")
-    history = body.get("history",[])
-    task    = body.get("task","*")
-    if not message: raise HTTPException(400,"Empty message")
-    accepted = ROLE_TASKS.get(WORKER_ROLE,["*"])
-    if "*" not in accepted and task not in accepted:
-        return JSONResponse({"ok":False,"reason":f"Worker [{WORKER_ROLE}] không xử lý task '{task}'"})
-    t0    = time.time()
-    reply = await process(message, system, history, task)
-    return JSONResponse({"ok":True,"reply":reply,"role":WORKER_ROLE,"elapsed":round(time.time()-t0,2)})
-
-@app.post("/api/run-code")
-async def run_code_endpoint(request: Request):
-    """Chạy code qua Piston (no key)."""
-    try: body = await request.json()
-    except Exception: raise HTTPException(400,"Invalid JSON")
-    code     = body.get("code","").strip()
-    lang     = body.get("lang","python")
-    stdin    = body.get("stdin","")
-    user_id  = body.get("user_id","unknown")
-    is_admin = body.get("is_admin",False)
-    # Admin check
-    if WORKER_SECRET and body.get("admin_secret","") == WORKER_SECRET:
-        is_admin = True
-    if not code: raise HTTPException(400,"Empty code")
-    ok, msg = _rate_ok(user_id, is_admin)
-    if not ok: return JSONResponse({"ok":False,"error":msg,"rate_blocked":True})
-    result = await run_code_piston(code, lang, stdin)
-    return JSONResponse(result)
-
-@app.get("/api/quota")
-async def quota_status(secret: str = ""):
-    import hmac as _hm2
-    if not WORKER_SECRET or not _hm2.compare_digest(secret.encode(), WORKER_SECRET.encode()):
-        raise HTTPException(403,"Unauthorized")
-    return {"role":WORKER_ROLE, "quota":{k:"cooldown" if not _quota_ok(k) else "ok" for k in _quota_hits},
-            "keys":{k:len(v) for k,v in _KEYS.items()}}
-
-
-# ── STEALTH PING ──────────────────────────────────────────────────
-_PING_RANGES = {
-    "code":     (240,360), "strong":   (270,420),
-    "creative": (300,480), "backup":   (330,540),
-}
-
-def _self_ping():
-    if not SELF_URL: return
-    time.sleep(random.randint(30,120))  # Stagger start
-    lo, hi = _PING_RANGES.get(WORKER_ROLE,(300,480))
-    while True:
-        try:
-            import requests as _rq
-            _rq.get(f"{SELF_URL}/health", timeout=10)
-            print(f"[WORKER-{WORKER_ROLE}] 💓 self-ping OK")
-        except Exception as e:
-            print(f"[WORKER-{WORKER_ROLE}] ⚠️ self-ping: {e}")
-        time.sleep(random.randint(lo, hi))
-
-def _cross_ping():
-    peers = [u for u in WORKER_URLS if u != SELF_URL]
-    if not peers: return
-    time.sleep(random.randint(120,300))
-    while True:
-        try:
-            import requests as _rq
-            _rq.get(f"{random.choice(peers)}/health", timeout=8)
-        except Exception: pass
-        time.sleep(random.randint(600,1200))
+    # Background: refresh keys mỗi 25 phút
+    async def _refresh_loop():
+        while True:
+            await asyncio.sleep(1500)
+            _fetch_keys_sync()
 
-if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+    asyncio.create_task(_refresh_loop())
+    print(f"[WORKER-{WORKER_ROLE.upper()}] ✅ Sẵn sàng! Role={WORKER_ROLE}")

requirements.txt

@@ -1,19 +1,9 @@
-# Worker space requirements — chỉ những gì worker cần
-# (nhẹ hơn main space, không cần gradio/playwright/maigret/...)
-httpx
 fastapi
 uvicorn
+httpx
 requests
-aiohttp
 openai
 cohere
+edge-tts
 google-generativeai>=0.8.3
-sentence-transformers
-faiss-cpu
-numpy
-apscheduler
 psutil
-python-multipart
-pydub
-websockets>=12.0
-pytz