# ================================================================ # URAL AI - SELF EVOLUTION ENGINE v2.0 # File: ural_self_evolution.py # CACH DUNG: import ural_self_evolution (cuoi cung trong app.py) # ================================================================ import os, sys, re, json, time, random, datetime, threading, traceback, subprocess from typing import Optional, Tuple # ── CONFIG ────────────────────────────────────────────────────── EVOLUTION_LOG_FILE = "ural_evolution_log.json" MODULES_DIR = "ural_modules" CODE_BRAIN = { "platform": "github", "base_url": "https://models.inference.ai.azure.com", "models": ["DeepSeek-R1", "gpt-4o", "DeepSeek-V3-0324"], "locked": True, } # ── FREE CODE BRAIN: Mistral → Gemini → Cerebras ──────────────────────────── # Khi GitHub (luồng 1) hết quota → tự động chạy theo thứ tự dưới # LUỒNG 2 = Mistral Codestral (code chuyên sâu, 1 tỷ tokens/tháng FREE) # LUỒNG 3 = Gemini (1500 req/ngày FREE) # LUỒNG 4 = Cerebras (backup cuối, nhanh) # Groq đã bỏ khỏi đây (bn không dùng Groq) CODE_BRAIN_FREE_FALLBACKS = [ # LUỒNG 2: MISTRAL — Codestral chuyên code, 1 tỷ tokens/tháng FREE # Dùng cho: auto-fix lỗi phức tạp, auto-upgrade module, auto-evaluate trước khi dùng { "platform": "mistral", "base_url": "https://api.mistral.ai/v1", "env_prefix": "MISTRAL_KEY", # Điền MISTRAL_KEY1, MISTRAL_KEY2... "models": [ "codestral-latest", # Chuyên code — ưu tiên nhất "mistral-large-latest", # Reasoning tốt — fallback "mistral-small-latest", # Backup nhẹ ], "daily_limit": 99999, # 1 tỷ tokens/tháng "signup_url": "https://console.mistral.ai", "note": "FREE, không thẻ, 1 tỷ tokens/tháng - tốt nhất cho code", }, # LUỒNG 3: GEMINI — 1500 req/ngày FREE (Google AI Studio) { "platform": "gemini", "base_url": "https://generativelanguage.googleapis.com/v1beta/openai", "env_prefix": "GEMINI_API_KEY", # Điền GEMINI_API_KEY1, GEMINI_API_KEY2... "models": [ "gemini-2.5-flash", # Mạnh nhất, ưu tiên "gemini-2.0-flash", # Nhanh, ổn định "gemini-1.5-flash", # Backup cuối ], "daily_limit": 1500, "signup_url": "https://aistudio.google.com/apikey", "note": "FREE, không thẻ, dùng nhiều Google account lấy nhiều key", }, # LUỒNG 4: CEREBRAS — backup nhanh khi Mistral + Gemini hết # ⚠️ (2026-07-17) Cerebras free tier ĐÓNG CỬA 17/08/2026 — bắt buộc thẻ # thanh toán từ ngày đó. Giữ khai báo để hunt logic tự phát hiện khi # Cerebras ngừng hoạt động (daily_limit/note cập nhật để phản ánh đúng # tình trạng); KHÔNG xóa hẳn để tránh phá cấu trúc _FLOWS mà module # khác có thể đang dò theo index/platform name. { "platform": "cerebras", "base_url": "https://api.cerebras.ai/v1", "env_prefix": "CEREBRAS_KEY", "models": [ "llama-3.3-70b", "llama3.1-70b", ], "daily_limit": 10000, "signup_url": "https://cloud.cerebras.ai", "note": "⚠️ Free tier đóng cửa 17/08/2026 (cần thẻ) — dùng yui_model_router làm nguồn chính từ nay", }, ] ADMIN_CHAT_BRAIN = {"platform": "cohere", "models": ["command-r-plus"], "locked": True} GUEST_BRAIN = { "platform": "cloudflare", "current_model": "@cf/meta/llama-3.1-8b-instruct", "candidates": [ "@cf/meta/llama-4-scout-17b-16e-instruct", "@cf/mistralai/mistral-small-3.1-24b-instruct", "@cf/google/gemma-3-12b-it", "@cf/meta/llama-3.3-70b-instruct-fp8-fast", ], "locked": False, } BACKUP_BRAIN = {"platform": "openrouter", "live_models": [], "locked": False} FINAL_SAFETY = {"platform": "cohere", "model": "command-r", "locked": True} CONFIG = { "max_upgrades_per_day": 1, "hunt_cooldown": 300, "benchmark_min_score": 6, "deepfix_hourly_limit": 3, "lab_timeout": 10, "lab_max_output": 2048, "lab_hourly_limit": 20, } HF_LIMITS = { "ram_max_pct": 75, "ram_danger_pct": 85, "cpu_max_pct": 70, "disk_min_free_gb": 1.0, "module_max_count": 10, "heavy_imports": [ "torch","tensorflow","keras","transformers", "scipy","sklearn","cv2","faiss", "diffusers","accelerate","bitsandbytes", ], } SAFE_IMPORTS = { "os","re","json","time","datetime","random", "math","string","collections","typing","requests", } BANNED_PATTERNS = [ r"exec\s*\(", r"eval\s*\(", r"__import__\s*\(", r"subprocess", r"os\.system", r"os\.popen", r"shutil\.rmtree", r"ural_self_evolution", ] # ================================================================ # EVOLUTION LOG # ================================================================ class EvolutionLog: def __init__(self): self._log = [] try: if os.path.exists(EVOLUTION_LOG_FILE): with open(EVOLUTION_LOG_FILE, "r", encoding="utf-8") as f: self._log = json.load(f) except Exception: self._log = [] def _save(self): try: with open(EVOLUTION_LOG_FILE, "w", encoding="utf-8") as f: json.dump(self._log[-500:], f, ensure_ascii=False, indent=2) except Exception: pass def record(self, event_type: str, detail: str, success: bool = True): entry = { "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "type": event_type, "detail": detail[:300], "success": success, } self._log.append(entry) self._save() print(f"[Evo/{'OK' if success else 'FAIL'}] {event_type}: {detail[:80]}") def get_recent(self, n: int = 10) -> list: return self._log[-n:] def count_today(self, event_type: str) -> int: today = datetime.datetime.now().strftime("%Y-%m-%d") return sum(1 for e in self._log if e.get("type") == event_type and e.get("time", "").startswith(today)) def format_recent(self, n: int = 10) -> str: recent = self.get_recent(n) if not recent: return "Chua co event nao." lines = [f"Evolution Log ({n} gan nhat):\n"] for e in reversed(recent): icon = "v" if e["success"] else "x" lines.append(f"[{icon}] {e['time']} [{e['type']}] {e['detail'][:80]}") return "\n".join(lines) _evo_log = EvolutionLog() # ================================================================ # HEALTH MONITOR # ================================================================ class PlatformHealthMonitor: def __init__(self): self.status = { p: {"alive": True, "fail_count": 0, "last_fail": ""} for p in ["cloudflare","cohere","github", "openrouter","huggingface","sambanova"] } self._lock = threading.Lock() def mark_fail(self, platform: str, error: str = ""): with self._lock: if platform not in self.status: return s = self.status[platform] s["fail_count"] += 1 s["last_fail"] = error[:100] if s["fail_count"] >= 3: s["alive"] = False _evo_log.record("platform_down", f"{platform}: {error[:60]}", False) def mark_success(self, platform: str): with self._lock: if platform in self.status: self.status[platform].update({"alive": True, "fail_count": 0}) def is_alive(self, platform: str) -> bool: return self.status.get(platform, {}).get("alive", True) def get_alive(self) -> list: return [p for p, s in self.status.items() if s["alive"]] def get_report(self) -> str: roles = { "github": "LOCKED CODE", "cohere": "LOCKED ADMIN+SAFETY", "cloudflare": "HUNTABLE GUEST", "openrouter": "HUNTABLE BACKUP", "huggingface": "ADMIN FALLBACK", "sambanova": "BACKUP", } lines = ["Health Monitor -- 6 nen tang:\n"] for p, s in self.status.items(): icon = "LIVE" if s["alive"] else "DOWN" lines.append( f" [{icon}] {p} [{roles.get(p,p)}] | Loi: {s['fail_count']}" + (f" | {s['last_fail'][:40]}" if s["last_fail"] else "") ) return "\n".join(lines) _health = PlatformHealthMonitor() # ================================================================ # HF RESOURCE GUARD # ================================================================ class HFResourceGuard: def __init__(self): self._last_check = 0 self._cache_ttl = 10 self._cached = None def get_resources(self) -> dict: now = time.time() if self._cached and now - self._last_check < self._cache_ttl: return self._cached try: import psutil mem = psutil.virtual_memory() cpu = psutil.cpu_percent(interval=0.5) disk = psutil.disk_usage("/") status = { "ram_total_gb": round(mem.total / 1024**3, 2), "ram_used_gb": round(mem.used / 1024**3, 2), "ram_pct": mem.percent, "cpu_pct": cpu, "disk_free_gb": round(disk.free / 1024**3, 2), "disk_pct": disk.percent, "ok": True, } except Exception as e: status = { "ram_total_gb": 16.0, "ram_used_gb": 8.0, "ram_pct": 50.0, "cpu_pct": 30.0, "disk_free_gb": 40.0, "disk_pct": 20.0, "ok": False, "error": str(e), } self._cached = status self._last_check = now return status def check_upgrade(self, code: str = "") -> Tuple[bool, str]: res = self.get_resources() issues = [] if res["ram_pct"] >= HF_LIMITS["ram_danger_pct"]: issues.append(f"RAM qua cao: {res['ram_pct']:.0f}%") if res["cpu_pct"] >= HF_LIMITS["cpu_max_pct"]: issues.append(f"CPU qua cao: {res['cpu_pct']:.0f}%") if res["disk_free_gb"] < HF_LIMITS["disk_min_free_gb"]: issues.append(f"Disk gan day: con {res['disk_free_gb']:.2f}GB") try: mc = len([f for f in os.listdir(MODULES_DIR) if f.endswith(".py") and f.startswith("auto_") ]) if os.path.exists(MODULES_DIR) else 0 if mc >= HF_LIMITS["module_max_count"]: issues.append(f"Qua nhieu modules: {mc}/{HF_LIMITS['module_max_count']}") except Exception: pass if code: for lib in HF_LIMITS["heavy_imports"]: pat = r"(?:import|from)\s+" + re.escape(lib) + r"\b" if re.search(pat, code): issues.append(f"Import thu vien nang: {lib}") if issues: return False, "CHAN:\n" + "\n".join(f" - {i}" for i in issues) warn = [] if res["ram_pct"] >= HF_LIMITS["ram_max_pct"]: warn.append(f"RAM cao: {res['ram_pct']:.0f}%") if warn: return True, "CANH BAO: " + ", ".join(warn) return True, "OK" def check_hunt(self) -> Tuple[bool, str]: res = self.get_resources() if res["cpu_pct"] >= 80: return False, f"CPU {res['cpu_pct']:.0f}%" if res["ram_pct"] >= HF_LIMITS["ram_danger_pct"]: return False, f"RAM {res['ram_pct']:.0f}%" return True, "OK" def dashboard(self) -> str: res = self.get_resources() safe, msg = self.check_upgrade() mc = 0 try: if os.path.exists(MODULES_DIR): mc = len([f for f in os.listdir(MODULES_DIR) if f.endswith(".py") and f.startswith("auto_")]) except Exception: pass lines = [ "HF Space Resource Dashboard:\n", f"RAM: {res['ram_used_gb']:.1f}/{res['ram_total_gb']:.1f}GB ({res['ram_pct']:.0f}%)", f"CPU: {res['cpu_pct']:.0f}%", f"Disk: con {res['disk_free_gb']:.1f}GB", f"Modules: {mc}/{HF_LIMITS['module_max_count']}", f"\nTrang thai: {'AN TOAN' if safe else 'CHUA AN TOAN'}", ] if not safe: lines.append(f"Ly do: {msg}") return "\n".join(lines) def cleanup_modules(self, keep: int = 5) -> str: if not os.path.exists(MODULES_DIR): return "Chua co module nao." try: files = sorted([ os.path.join(MODULES_DIR, f) for f in os.listdir(MODULES_DIR) if f.endswith(".py") and f.startswith("auto_") ]) to_del = files[:-keep] if len(files) > keep else [] for f in to_del: os.remove(f) _evo_log.record("cleanup", f"Xoa {len(to_del)} module cu", True) return f"Da xoa {len(to_del)} module cu, giu lai {len(files)-len(to_del)}." except Exception as e: return f"Loi: {e}" _resource_guard = HFResourceGuard() # ================================================================ # GITHUB QUOTA GUARD - Bảo vệ key GitHub khỏi bị đốt # ================================================================ class GitHubQuotaGuard: """ Vấn đề: Module quét lỗ hổng gọi DeepSeek-R1 hàng loạt = đốt hết key. Giải pháp: Đếm số lần gọi, chặn khi gần giới hạn, auto-fallback sang Mistral/Gemini/Cerebras. """ DAILY_LIMIT = 40 HOURLY_LIMIT = 8 VULN_LIMIT = 2 def __init__(self): self._daily_calls = [] self._hourly_calls = [] self._vuln_calls = [] self._lock = threading.Lock() def _clean(self): now = time.time() self._daily_calls = [t for t in self._daily_calls if now - t < 86400] self._hourly_calls = [t for t in self._hourly_calls if now - t < 3600] self._vuln_calls = [t for t in self._vuln_calls if now - t < 86400] def can_call(self, call_type: str = "general") -> tuple: with self._lock: self._clean() if len(self._daily_calls) >= self.DAILY_LIMIT: h = (86400 - (time.time() - self._daily_calls[0])) / 3600 return False, f"GitHub quota ngay ({self.DAILY_LIMIT}). Reset sau {h:.1f}h" if len(self._hourly_calls) >= self.HOURLY_LIMIT: m = (3600 - (time.time() - self._hourly_calls[0])) / 60 return False, f"GitHub rate limit ({self.HOURLY_LIMIT}/gio). Thu lai sau {m:.0f}m" if call_type == "vuln_scan" and len(self._vuln_calls) >= self.VULN_LIMIT: return False, f"Module quet lo hong da dung {self.VULN_LIMIT} lan hom nay" return True, "OK" def record_call(self, call_type: str = "general"): with self._lock: now = time.time() self._daily_calls.append(now) self._hourly_calls.append(now) if call_type == "vuln_scan": self._vuln_calls.append(now) def status(self) -> str: with self._lock: self._clean() return ( f"GitHub Quota Guard:\n" f" Hom nay: {len(self._daily_calls)}/{self.DAILY_LIMIT}\n" f" Gio nay: {len(self._hourly_calls)}/{self.HOURLY_LIMIT}\n" f" Quet lo hong: {len(self._vuln_calls)}/{self.VULN_LIMIT}/ngay" ) _github_quota = GitHubQuotaGuard() def _get_free_keys(env_prefix: str) -> list: """Quet bien moi truong lay key theo prefix (VD: MISTRAL_KEY1, MISTRAL_KEY2...)""" keys = [] for k, v in os.environ.items(): if k.startswith(env_prefix) and v.strip(): keys.append(v.strip()) base = os.getenv(env_prefix) if base and base.strip() and base.strip() not in keys: keys.append(base.strip()) return keys def _call_free_code_brain(prompt: str) -> Optional[str]: """ Goi CODE BRAIN mien phi khi GitHub (luong 1) het quota: Luong 2: Mistral Codestral (chuyen code, 1 ty tokens/thang) Luong 3: Gemini 2.5/2.0 Flash (1500 req/ngay per project) Luong 4: Cerebras (backup nhanh, free) DANG KY: Tat ca deu FREE, KHONG THE, KHONG PHI AN: - Mistral: https://console.mistral.ai - Gemini: https://aistudio.google.com/apikey - Cerebras: https://cloud.cerebras.ai """ import openai as _oa for fb in CODE_BRAIN_FREE_FALLBACKS: keys = _get_free_keys(fb["env_prefix"]) if not keys: continue key = random.choice(keys) for model in fb["models"]: try: if fb["platform"] == "mistral": try: import mistralai as _mst mc = _mst.Mistral(api_key=key) resp = mc.chat.complete( model=model, messages=[ {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, {"role": "user", "content": prompt}, ], max_tokens=3000, temperature=0.1, ) raw = resp.choices[0].message.content.strip() except ImportError: # Fallback dung openai compat client = _oa.OpenAI(api_key=key, base_url=fb["base_url"]) r = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, {"role": "user", "content": prompt}, ], max_tokens=3000, temperature=0.1, ) raw = r.choices[0].message.content.strip() else: client = _oa.OpenAI(api_key=key, base_url=fb["base_url"]) r = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, {"role": "user", "content": prompt}, ], max_tokens=3000, temperature=0.1, ) raw = r.choices[0].message.content.strip() raw = re.sub(r"^```python\s*", "", raw) raw = re.sub(r"^```\s*", "", raw) raw = re.sub(r"\s*```$", "", raw) print(f"[FREE_CODE_BRAIN] {fb['platform']}/{model}") _evo_log.record("code_gen_free", f"{fb['platform']}/{model}", True) return raw.strip() except Exception as e: print(f" [{fb['platform']}/{model}] loi: {str(e)[:60]}") continue return None # ================================================================ # CODE LAB ENGINE -- Phong thi nghiem chay code that # ================================================================ class CodeLabEngine: """Chay code trong subprocess co lap -- THAT SU, khong chi verify tinh.""" def __init__(self): self._run_times = [] self._lab_history = [] def _check_quota(self) -> Tuple[bool, str]: now = time.time() self._run_times = [t for t in self._run_times if now - t < 3600] if len(self._run_times) >= CONFIG["lab_hourly_limit"]: return False, f"Lab het quota ({CONFIG['lab_hourly_limit']} lan/gio)" return True, "OK" def run(self, code: str, description: str = "") -> dict: """Chay code trong subprocess co lap.""" ok, msg = self._check_quota() if not ok: return {"success": False, "output": "", "error": msg, "runtime_ms": 0, "ran": False} res = _resource_guard.get_resources() if res["cpu_pct"] > 85: return {"success": False, "output": "", "error": f"CPU {res['cpu_pct']:.0f}% -- lab tam hoan", "runtime_ms": 0, "ran": False} start = time.time() try: proc = subprocess.run( [sys.executable, "-c", code], capture_output=True, text=True, timeout=CONFIG["lab_timeout"], env={**os.environ, "PYTHONPATH": ""}, ) rt = int((time.time() - start) * 1000) out = proc.stdout[:CONFIG["lab_max_output"]] err = proc.stderr[:CONFIG["lab_max_output"]] ok2 = (proc.returncode == 0) result = { "success": ok2, "output": out, "error": err if not ok2 else "", "returncode": proc.returncode, "runtime_ms": rt, "ran": True, "description": description, "time": datetime.datetime.now().strftime("%H:%M:%S"), } except subprocess.TimeoutExpired: result = { "success": False, "output": "", "error": f"TIMEOUT: qua {CONFIG['lab_timeout']}s", "returncode": -1, "runtime_ms": CONFIG["lab_timeout"] * 1000, "ran": True, "description": description, "time": datetime.datetime.now().strftime("%H:%M:%S"), } except Exception as e: result = { "success": False, "output": "", "error": f"LAB ERROR: {e}", "returncode": -1, "runtime_ms": 0, "ran": False, "description": description, "time": datetime.datetime.now().strftime("%H:%M:%S"), } self._run_times.append(time.time()) self._lab_history.append(result) if len(self._lab_history) > 50: self._lab_history.pop(0) _evo_log.record("lab_run", f"{'OK' if result['success'] else 'FAIL'}" f" ({result['runtime_ms']}ms): {description[:50]}", result["success"]) return result def test_module(self, code: str, runs: int = 3) -> dict: """ Test module NHIEU LAN de dam bao ket qua nhat quan. Quy trinh: 1. Import test x{runs} lan -- tat ca phai pass 2. Main() test x{runs} lan -- output phai GIONG NHAU ca {runs} lan 3. Chi pass neu: tat ca runs pass VA output nhat quan Tai sao 3 lan: - Tranh truong hop code co randomness che giau bug - Tranh truong hop may man pass 1 lan nhung thuc ra co race condition - Tranh code phu thuoc state ben ngoai (lan 1 ok, lan 2 fail) """ n = max(1, min(runs, 5)) # gioi han 1-5 lan # Pha 1: Import test x n lan import_results = [] for i in range(n): r = self.run(code, f"import test {i+1}/{n}") import_results.append(r) if not r["success"]: # Fail ngay -- khong can chay tiep return { "import_ok": False, "import_error": r.get("error", ""), "main_ok": None, "main_output": "", "main_error": "", "overall_ok": False, "runs_done": i + 1, "consistent": False, "fail_reason": f"Import fail lan {i+1}/{n}: {r.get('error','')[:80]}", } # Pha 2: Main test x n lan (neu co __name__ block) main_results = [] if "__name__" in code: test_code = code.replace( "if __name__ == '__main__':", "if True: # LAB TRIPLE-RUN" ).replace( 'if __name__ == "__main__":', "if True: # LAB TRIPLE-RUN" ) for i in range(n): r = self.run(test_code, f"main test {i+1}/{n}") main_results.append(r) if not r["success"]: return { "import_ok": True, "import_error": "", "main_ok": False, "main_output": "", "main_error": r.get("error", ""), "overall_ok": False, "runs_done": n + i + 1, "consistent": False, "fail_reason": f"Main fail lan {i+1}/{n}: {r.get('error','')[:80]}", } # Pha 3: Kiem tra tinh nhat quan output consistent = True consistency_note = "" if main_results: outputs = [r.get("output", "").strip() for r in main_results] if len(set(outputs)) > 1: consistent = False consistency_note = ( f"Output KHONG nhat quan qua {n} lan chay:\n" + "\n".join(f" Lan {i+1}: {o[:60]}" for i, o in enumerate(outputs)) ) overall_ok = ( all(r["success"] for r in import_results) and (not main_results or all(r["success"] for r in main_results)) and consistent ) best_output = main_results[-1].get("output", "") if main_results else "" avg_runtime = ( sum(r["runtime_ms"] for r in import_results + main_results) // max(1, len(import_results) + len(main_results)) ) return { "import_ok": True, "import_error": "", "main_ok": bool(main_results) and all(r["success"] for r in main_results), "main_output": best_output, "main_error": "", "overall_ok": overall_ok, "runs_done": len(import_results) + len(main_results), "consistent": consistent, "consistency_note":consistency_note, "avg_runtime_ms": avg_runtime, "fail_reason": consistency_note if not consistent else "", } def test_fix(self, original: str, fixed: str, runs: int = 3) -> dict: """ So sanh code goc vs code da fix, chay {runs} lan. Fix chi duoc chap nhan neu: 1. Code goc fail (hoac fix tot hon ro rang) 2. Code da fix PASS CA {runs} LAN 3. Output nhat quan qua {runs} lan """ n = max(1, min(runs, 5)) # Code goc -- chay 1 lan de confirm no fail orig_r = self.run(original, "original") # Code fix -- chay n lan, phai pass het fix_results = [] for i in range(n): r = self.run(fixed, f"fix run {i+1}/{n}") fix_results.append(r) all_pass = all(r["success"] for r in fix_results) outputs = [r.get("output", "").strip() for r in fix_results] consistent = (len(set(outputs)) <= 1) fix_is_good = all_pass and consistent return { "original_failed": not orig_r["success"], "fix_passed": all_pass, "fix_is_better": (fix_is_good and not orig_r["success"]), "fix_consistent": consistent, "runs_done": n, "pass_count": sum(1 for r in fix_results if r["success"]), "original_error": orig_r.get("error", "")[:200], "fix_output": fix_results[-1].get("output", "")[:200] if fix_results else "", "inconsistency": "" if consistent else f"Output khac nhau qua {n} lan", } def history(self, n: int = 5) -> str: if not self._lab_history: return "Lab chua chay lan nao." lines = [f"Lab History ({n} gan nhat):\n"] for r in reversed(self._lab_history[-n:]): icon = "v" if r["success"] else "x" lines.append( f"[{icon}] {r['time']} ({r['runtime_ms']}ms)" f" -- {r['description'][:50]}" ) if not r["success"] and r.get("error"): lines.append(f" {r['error'][:60]}") used = len(self._run_times) lines.append(f"\nQuota: {used}/{CONFIG['lab_hourly_limit']} lan/gio") return "\n".join(lines) _lab = CodeLabEngine() # ================================================================ # BENCHMARK CHAT QUALITY -- Danh gia chat luong output AI # ================================================================ class BenchmarkQuality: """ Danh gia chat luong output cua module/fix truoc khi ap dung. Test 3 tieu chi: 1. Correctness -- output co dung voi expected khong (0-10) 2. Performance -- thoi gian chay co chap nhan duoc khong (0-10) 3. Stability -- co crash hay exception an khong (0-10) Ket qua: score tong 0-30. Pass neu >= PASS_THRESHOLD. """ PASS_THRESHOLD = 20 # /30 -- phai dat it nhat 20/30 moi duoc dung def score_module(self, code: str, expected_outputs: list = None, max_runtime_ms: int = 3000) -> dict: """ Benchmark module voi 3 tieu chi. expected_outputs: list cac chuoi phai co trong output (neu co) max_runtime_ms: thoi gian toi da chap nhan (ms) """ # Chay 3 lan lay ket qua lab_result = _lab.test_module(code, runs=3) scores = {} details = [] # Tieu chi 1: Correctness (0-10) correctness = 0 if lab_result["overall_ok"]: correctness += 6 # Base: code chay duoc if lab_result.get("consistent"): correctness += 2 # Output nhat quan if expected_outputs and lab_result.get("main_output"): output = lab_result["main_output"].lower() matched = sum(1 for e in expected_outputs if str(e).lower() in output) ratio = matched / len(expected_outputs) correctness += int(ratio * 2) # 0-2 diem them else: correctness += 2 # Khong co expected = khong tru diem scores["correctness"] = min(10, correctness) details.append(f"Correctness: {scores['correctness']}/10" + (" (output nhat quan)" if lab_result.get("consistent") else " (output KHONG nhat quan)")) # Tieu chi 2: Performance (0-10) perf = 10 avg_rt = lab_result.get("avg_runtime_ms", 0) if avg_rt > max_runtime_ms: perf -= 5 # Qua cham elif avg_rt > max_runtime_ms * 0.7: perf -= 2 # Hoi cham if not lab_result.get("ran", True): perf = 0 # Khong chay duoc scores["performance"] = max(0, perf) details.append(f"Performance: {scores['performance']}/10" + f" (avg {avg_rt}ms, limit {max_runtime_ms}ms)") # Tieu chi 3: Stability (0-10) stability = 10 if not lab_result["import_ok"]: stability = 0 # Loi import = fail hoan toan elif not lab_result.get("overall_ok"): stability = 4 # Co loi khi chay elif lab_result.get("fail_reason"): stability = 6 # Co van de nhung khong crash scores["stability"] = stability details.append(f"Stability: {scores['stability']}/10" + ("" if stability == 10 else f" ({lab_result.get('fail_reason','')[:40]})")) total = sum(scores.values()) passed = total >= self.PASS_THRESHOLD return { "passed": passed, "total_score": total, "max_score": 30, "pass_threshold": self.PASS_THRESHOLD, "scores": scores, "details": details, "lab_result": lab_result, "verdict": ( f"PASS ({total}/30)" if passed else f"FAIL ({total}/30 -- can {self.PASS_THRESHOLD})" ), } def format_report(self, bm: dict) -> str: lines = [ f"Benchmark Quality Report:", f" Ket qua: {bm['verdict']}", ] for d in bm.get("details", []): lines.append(f" - {d}") if not bm["passed"]: lines.append( f" --> KHONG dat nguong {bm['pass_threshold']}/30" f" -- module bi CHON KHONG duoc dung" ) return "\n".join(lines) _bm_quality = BenchmarkQuality() # ================================================================ # ROLLBACK MANAGER -- Tu dong rollback neu module moi bi loi # ================================================================ class RollbackManager: """ Quan ly rollback tu dong khi module moi bi loi sau khi active. Cach hoat dong: 1. Truoc khi luu module moi --> tao snapshot trang thai hien tai 2. Neu module moi bi loi sau khi active --> tu dong rollback 3. Log day du moi thao tac 4. Giu toi da 5 snapshot (don dep tu dong) An toan: - Chi rollback module trong ural_modules/ (KHONG rollback app goc) - Rollback = xoa file module moi + phuc hoi snapshot - Neu rollback cung fail --> bao admin qua log """ MAX_SNAPSHOTS = 5 def __init__(self): self.snapshots = [] # [{name, file, backup_content, time}] self.rollback_log = [] def create_snapshot(self, module_name: str, file_path: str) -> bool: """Tao snapshot truoc khi luu module moi.""" try: # Doc noi dung hien tai neu file ton tai backup = None if os.path.exists(file_path): with open(file_path, "r", encoding="utf-8") as f: backup = f.read() snapshot = { "name": module_name, "file": file_path, "backup_content": backup, "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "exists_before": os.path.exists(file_path), } self.snapshots.append(snapshot) # Don dep snapshot cu if len(self.snapshots) > self.MAX_SNAPSHOTS: self.snapshots.pop(0) _evo_log.record("snapshot", f"Tao snapshot: {module_name}", True) return True except Exception as e: _evo_log.record("snapshot", f"Loi tao snapshot: {e}", False) return False def rollback(self, module_name: str, reason: str = "") -> str: """ Tu dong rollback module ve trang thai truoc do. Goi khi phat hien module moi bi loi. """ snapshot = next( (s for s in reversed(self.snapshots) if s["name"] == module_name), None, ) if not snapshot: msg = f"Khong tim thay snapshot cho: {module_name}" _evo_log.record("rollback", msg, False) return msg file_path = snapshot["file"] try: if snapshot["exists_before"] and snapshot["backup_content"]: # Phuc hoi noi dung cu with open(file_path, "w", encoding="utf-8") as f: f.write(snapshot["backup_content"]) msg = f"Da rollback {module_name} ve trang thai truoc." else: # File nay la moi -- xoa di if os.path.exists(file_path): os.remove(file_path) msg = f"Da xoa module moi {module_name} (rollback)." self.rollback_log.append({ "module": module_name, "reason": reason[:100], "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "ok": True, }) _evo_log.record("rollback", f"OK: {module_name} -- {reason[:60]}", True) return msg except Exception as e: err_msg = f"Loi rollback {module_name}: {e}" self.rollback_log.append({ "module": module_name, "reason": reason[:100], "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "ok": False, "error": str(e), }) _evo_log.record("rollback", err_msg, False) return err_msg def auto_check_and_rollback(self, module_name: str, file_path: str) -> Optional[str]: """ Sau khi luu module moi, chay kiem tra lan cuoi. Neu fail --> tu dong rollback, khong can admin. Tra ve None neu OK, hoac error message neu rollback. """ if not os.path.exists(file_path): return None try: with open(file_path, "r", encoding="utf-8") as f: code = f.read() except Exception as e: return self.rollback(module_name, f"Khong doc duoc file: {e}") # Chay kiem tra cuoi: 3 lan, tat ca phai pass final_check = _lab.test_module(code, runs=3) if not final_check["overall_ok"] or not final_check.get("consistent", True): reason = final_check.get("fail_reason", "Lab test fail sau khi luu") print(f"[Rollback] Module {module_name} fail post-save check!") print(f" Ly do: {reason}") msg = self.rollback(module_name, reason) return f"AUTO ROLLBACK: {msg}\nLy do: {reason}" _evo_log.record("post_save_check", f"{module_name}: OK", True) return None # Khong can rollback def get_log(self) -> str: if not self.rollback_log: return "Chua co rollback nao." lines = ["Rollback Log:\n"] for r in reversed(self.rollback_log[-5:]): icon = "v" if r["ok"] else "x" lines.append( f"[{icon}] {r['time']} -- {r['module']}: {r['reason'][:60]}" ) return "\n".join(lines) _rollback = RollbackManager() # ================================================================ # MODEL BENCHMARKER # ================================================================ class ModelBenchmarker: TESTS = [ ("Thu do Viet Nam la gi?", "ha noi"), ("1 + 1 bang may?", "2"), ("Python la gi?", "lap trinh"), ("Mat trang quay quanh gi?", "trai dat"), ] def quick_check(self, model_id: str, platform: str, keys: list, base_url: str = None) -> bool: if not keys: return False try: key = random.choice(keys) if platform == "cohere": import cohere as _co co = _co.Client(api_key=key) r = co.chat(message="Hi", model=model_id) return bool(r.text) else: import openai as _oa url = base_url or "https://openrouter.ai/api/v1" client = _oa.OpenAI(api_key=key, base_url=url) extra = {} if "openrouter" in url: extra = {"extra_headers": { "HTTP-Referer": "https://huggingface.co", "X-Title": "URAL AI", }} r = client.chat.completions.create( model=model_id, messages=[{"role": "user", "content": "Hi"}], max_tokens=5, **extra, ) return bool(r.choices[0].message.content) except Exception: return False def benchmark(self, model_id: str, platform: str, keys: list, base_url: str = None) -> dict: if not keys: return {"score": 0, "max": 8, "passed": False} key = random.choice(keys) score = 0 for question, expected in self.TESTS: try: if platform == "cohere": import cohere as _co co = _co.Client(api_key=key) resp = co.chat(message=question, model=model_id) ans = resp.text.lower() else: import openai as _oa url = base_url or "https://openrouter.ai/api/v1" client = _oa.OpenAI(api_key=key, base_url=url) extra = {} if "openrouter" in url: extra = {"extra_headers": { "HTTP-Referer": "https://huggingface.co", "X-Title": "URAL AI", }} r = client.chat.completions.create( model=model_id, messages=[{"role": "user", "content": question}], max_tokens=50, **extra, ) ans = (r.choices[0].message.content or "").lower() if expected.lower() in ans: score += 2 except Exception: pass passed = score >= CONFIG["benchmark_min_score"] return {"score": score, "max": len(self.TESTS) * 2, "passed": passed} _bench = ModelBenchmarker() # ================================================================ # MODEL HUNTER v2 -- Chi hunt Tang 3 & 4 # ================================================================ class ModelHunterV2: LOCKED = {"github", "cohere_plus"} def __init__(self): self.cf_alive = [] self.or_alive = [] self.dead = set() self.last_cf = 0 self.last_or = 0 self._lock = threading.Lock() def _app(self): for mod in sys.modules.values(): if hasattr(mod, "CLOUDFLARE_KEYS") and hasattr(mod, "COHERE_KEYS"): return mod return None def hunt_all(self, force: bool = False): app = self._app() if not app: return safe, reason = _resource_guard.check_hunt() if not safe: print(f"[Hunter] Hoan hunt: {reason}") _evo_log.record("hunt_blocked", reason[:80], False) def _retry(): time.sleep(300) s2, _ = _resource_guard.check_hunt() if s2: self.hunt_all(force=force) threading.Thread(target=_retry, daemon=True).start() return cf_k = getattr(app, "CLOUDFLARE_KEYS", []) or_k = getattr(app, "OPENROUTER_KEYS", []) hf_t = getattr(app, "HF_TOKEN", None) if cf_k: threading.Thread(target=self._hunt_cf, args=(cf_k, force), daemon=True).start() if or_k: threading.Thread(target=self._hunt_or, args=(or_k, force), daemon=True).start() if hf_t: threading.Thread(target=self._check_hf, args=(hf_t,), daemon=True).start() def _hunt_cf(self, keys: list, force: bool = False): if not force and time.time() - self.last_cf < CONFIG["hunt_cooldown"]: return print("[Hunter-CF] Testing Cloudflare models...") found = [] for model in GUEST_BRAIN["candidates"]: if model in self.dead: continue key = random.choice(keys) if "|" not in key: continue acc_id, token = key.split("|", 1) cf_url = f"https://api.cloudflare.com/client/v4/accounts/{acc_id}/ai/v1" alive = _bench.quick_check(model, "cloudflare", [f"{acc_id}|{token}"], cf_url) if not alive: self.dead.add(model) continue bm = _bench.benchmark(model, "cloudflare", [f"{acc_id}|{token}"], cf_url) if bm["passed"]: found.append(model) print(f" LIVE: {model[:45]} score={bm['score']}/{bm['max']}") _evo_log.record("hunt_cf", f"Live: {model[:50]}", True) else: print(f" LOW: {model[:45]} score={bm['score']}/{bm['max']}") with self._lock: if found: self.cf_alive = found GUEST_BRAIN["current_model"] = found[0] app = self._app() if app: setattr(app, "_cf_current_model", found[0]) _health.mark_success("cloudflare") _evo_log.record("model_inject", f"CF -> {found[0]}", True) print(f"[Hunter-CF] Injected: {found[0]}") else: print("[Hunter-CF] Khong tim duoc model moi, giu nguyen") self.last_cf = time.time() def _hunt_or(self, keys: list, force: bool = False): if not force and time.time() - self.last_or < CONFIG["hunt_cooldown"]: return print("[Hunter-OR] Fetching OpenRouter free models...") try: import requests as _req resp = _req.get( "https://openrouter.ai/api/v1/models", headers={"Authorization": f"Bearer {keys[0]}"}, timeout=10, ) all_m = resp.json().get("data", []) free_ids = [ m["id"] for m in all_m if (str(m.get("pricing", {}).get("prompt", "1")) == "0" or m["id"].endswith(":free")) and m["id"] not in self.dead ] priority = ["deepseek", "gemma-3", "llama-3", "qwen", "mistral"] free_ids.sort(key=lambda mid: next( (i for i, kw in enumerate(priority) if kw in mid.lower()), 99 )) found = [] key = random.choice(keys) for mid in free_ids[:8]: if not _bench.quick_check(mid, "openrouter", [key]): self.dead.add(mid) continue bm = _bench.benchmark(mid, "openrouter", [key]) if bm["passed"]: found.append(mid) _evo_log.record("hunt_or", f"Live: {mid[:50]}", True) if len(found) >= 5: break with self._lock: if found: self.or_alive = found BACKUP_BRAIN["live_models"] = found app = self._app() if app: try: app.OPENROUTER_LIVE_MODELS = found except Exception: pass _health.mark_success("openrouter") _evo_log.record("model_inject", f"OR -> {len(found)} models", True) print(f"[Hunter-OR] Injected {len(found)} models") self.last_or = time.time() except Exception as e: _evo_log.record("hunt_or", f"Loi: {e}", False) def _check_hf(self, token: str): from huggingface_hub import InferenceClient alive = [] for model in ["Qwen/Qwen2.5-72B-Instruct", "mistralai/Mistral-7B-Instruct-v0.3"]: try: client = InferenceClient(model=model, token=token) r = client.chat_completion( messages=[{"role": "user", "content": "Hi"}], max_tokens=5 ) if r.choices[0].message.content: alive.append(model) print(f" [HF] LIVE: {model}") except Exception as e: _health.mark_fail("huggingface", str(e)[:60]) if alive: _health.mark_success("huggingface") ADMIN_CHAT_BRAIN["fallbacks"] = alive def status(self) -> str: lines = [ "Model Hunter v2 Status:\n", "Tang 1 CODE (LOCKED):", f" GitHub: {', '.join(CODE_BRAIN['models'])}", "\nTang 2 ADMIN CHAT (LOCKED):", f" Cohere: {', '.join(ADMIN_CHAT_BRAIN['models'])}", "\nTang 3 GUEST (Hunter quan ly):", f" CF: {GUEST_BRAIN['current_model']}", f" Pass benchmark: {len(self.cf_alive)} models", "\nTang 4 BACKUP (Hunter quan ly):", f" OR live: {len(self.or_alive)} models", f" {str(self.or_alive[:2]) if self.or_alive else '(chua hunt)'}", f"\nDead: {len(self.dead)}", ] return "\n".join(lines) _hunter = ModelHunterV2() # ================================================================ # SANDBOX VERIFY - 3 LOP CHONG BYPASS (tich hop truc tiep, khong can file ngoai) # Lap 1: AST Visitor - bat ham/import nguy hiem # Lap 2: Regex - bat trick ghep chuoi bypass # Lap 3: Compile check # Chong duoc: __builtins__['__import__'], getattr(__builtins__,'ev'+'al'), chr(101)+... # ================================================================ import ast as _ast_mod _SANDBOX_ALLOWED_IMPORTS = { "os","re","json","time","datetime","random","math","string", "collections","typing","requests","hashlib","base64","urllib", "functools","itertools","operator","copy", } _SANDBOX_BANNED_NAMES = { "eval","exec","compile","execfile","__import__","__builtins__", "__loader__","globals","locals","vars","dir","input","open", "subprocess","Popen","system","popen","spawn","fork","execv", "rmtree","remove","unlink","socket","pickle","marshal","ctypes", } _SANDBOX_BYPASS_PATTERNS = [ r"__builtins__\s*[\[\.]", r"getattr\s*\(\s*__", r"getattr\s*\(.*builtins", r"chr\s*\(\s*\d+\s*\)", r"\\x[0-9a-fA-F]{2}", r"importlib", r"__import__", r"os\.environ", r"os\.getenv", r"os\.remove\s*\(", r"os\.unlink\s*\(", r"shutil\.", r"\.write\s*\(", r"open\s*\(.*['\"][wa]['\"]", ] class _ASTSecurityVisitor(_ast_mod.NodeVisitor): def __init__(self): self.violations = [] def visit_Call(self, node): name = "" if isinstance(node.func, _ast_mod.Name): name = node.func.id elif isinstance(node.func, _ast_mod.Attribute): name = node.func.attr if name in _SANDBOX_BANNED_NAMES: self.violations.append(f"Ham bi cam: {name}() dong {getattr(node,'lineno',0)}") self.generic_visit(node) def visit_Import(self, node): for a in node.names: base = a.name.split(".")[0] if base not in _SANDBOX_ALLOWED_IMPORTS: self.violations.append(f"Import khong phep: {a.name}") self.generic_visit(node) def visit_ImportFrom(self, node): if node.module: base = node.module.split(".")[0] if base not in _SANDBOX_ALLOWED_IMPORTS: self.violations.append(f"Import khong phep: from {node.module}") self.generic_visit(node) def visit_Attribute(self, node): if isinstance(node.value, _ast_mod.Name): pair = f"{node.value.id}.{node.attr}" danger = {"os.system","os.popen","os.execv","os.remove","os.unlink", "os.environ","os.getenv","sys.exit","subprocess.Popen", "subprocess.call","shutil.rmtree"} if pair in danger: self.violations.append(f"Truy cap nguy hiem: {pair} dong {getattr(node,'lineno',0)}") self.generic_visit(node) def visit_Global(self, node): self.violations.append("Cam dung 'global' trong module tu dong") self.generic_visit(node) def _sandbox_verify(code: str) -> Tuple[bool, str]: """ Sandbox 3 lop: 1. Parse + AST Visitor (bat ham/import nguy hiem, chong bypass nang cao) 2. Regex bypass patterns (bat trick ghep chuoi, chr(), hex escape...) 3. Compile check """ if not code or not code.strip(): return False, "Code rong" if len(code) > 50000: return False, f"Code qua dai: {len(code)} ky tu" # Lop 1: AST try: tree = _ast_mod.parse(code) except SyntaxError as e: return False, f"Syntax error: {e}" except Exception as e: return False, f"Khong parse duoc: {e}" visitor = _ASTSecurityVisitor() try: visitor.visit(tree) except Exception as e: return False, f"Loi scan AST: {e}" if visitor.violations: return False, "Vi pham bao mat:\n" + "\n".join(f" - {v}" for v in visitor.violations[:5]) # Lop 2: Regex bypass for pat in _SANDBOX_BYPASS_PATTERNS: try: if re.search(pat, code, re.IGNORECASE | re.MULTILINE): return False, f"Pattern bypass bi chan: {pat[:50]}" except re.error: continue # Lop 3: Compile try: compile(code, "", "exec") except Exception as e: return False, f"Compile error: {e}" return True, "OK" def _call_code_brain(prompt: str, github_keys: list, call_type: str = "general") -> Optional[str]: """ Goi CODE_BRAIN voi 2 tang bao ve: 1. Kiem tra GitHubQuotaGuard truoc 2. Neu GitHub het quota → tu dong fallback sang Mistral/Gemini/Cerebras (tat ca FREE) Luon dung CODE_BRAIN cho code -- khong bao gio dung model chat thuong. """ # Tang 1: Thu GitHub (neu con quota) can_github, reason = _github_quota.can_call(call_type) if can_github and github_keys: key = random.choice(github_keys) for model in CODE_BRAIN["models"]: try: import openai as _oa client = _oa.OpenAI(api_key=key, base_url=CODE_BRAIN["base_url"]) resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "Expert Python developer. Chi tra ve code Python thuan."}, {"role": "user", "content": prompt}, ], max_tokens=3000, temperature=0.1, ) code = resp.choices[0].message.content.strip() code = re.sub(r"^```python\s*", "", code) code = re.sub(r"^```\s*", "", code) code = re.sub(r"\s*```$", "", code) # Ghi nhan da dung quota _github_quota.record_call(call_type) print(f"[CODE_BRAIN] GitHub/{model} | Quota hom nay: {len(_github_quota._daily_calls)}/{_github_quota.DAILY_LIMIT}") _evo_log.record("code_gen", f"GitHub/{model}", True) return code.strip() except Exception as e: print(f" {model} loi: {str(e)[:50]}") continue # GitHub loi het → fallback print(f"[CODE_BRAIN] GitHub that bai, chuyen sang FREE fallback...") else: print(f"[CODE_BRAIN] GitHub bi chan: {reason}") print(f"[CODE_BRAIN] Dang dung FREE fallback (Mistral/Gemini/Cerebras)...") # Tang 2: FREE fallback platforms result = _call_free_code_brain(prompt) if result: return result print("[CODE_BRAIN] Tat ca platforms deu het key hoac loi") return None def _get_github_keys() -> list: for mod in sys.modules.values(): if hasattr(mod, "GITHUB_KEYS"): return getattr(mod, "GITHUB_KEYS", []) return [] # ================================================================ # AUTO BUG FIXER -- 2 tang # ================================================================ class AutoBugFixer: """ Tang 1 Quick: quota/timeout/deprecated --> xu ly ngay, khong can AI Tang 2 Deep: loi phuc tap --> DeepSeek-R1 phan tich --> patch code Patch phai qua: Sandbox verify + Lab test + Admin confirm """ QUICK = { "quota": "switch_platform", "rate limit": "wait_retry", "429": "wait_retry", "connection": "mark_dead", "timeout": "mark_dead", "503": "mark_dead", "502": "mark_dead", "model not found": "hunt_replacement", "deprecated": "hunt_replacement", "invalid api key": "mark_key_bad", "authentication": "mark_key_bad", } DEEP_TRIGGERS = [ "typeerror","attributeerror","nameerror","keyerror", "valueerror","indexerror","syntaxerror","importerror", "json","parse","decode","unexpected", ] def __init__(self): self.quick_count = 0 self.deep_count = 0 self.fail_count = 0 self.error_log = [] self._deep_times = [] def _quick_fix(self, error: Exception, platform: str) -> Optional[str]: err_str = str(error).lower() action = next( (act for pat, act in self.QUICK.items() if pat in err_str), None ) if not action: return None if action == "mark_dead": _health.mark_fail(platform, str(error)[:80]) elif action == "switch_platform": _health.mark_fail(platform, "quota") elif action == "wait_retry": time.sleep(3) elif action == "hunt_replacement": if platform in ("cloudflare", "openrouter"): threading.Thread( target=_hunter.hunt_all, kwargs={"force": True}, daemon=True ).start() self.quick_count += 1 self._log(platform, str(error), action, "quick") _evo_log.record("bug_fix_quick", f"[{platform}] {action}: {str(error)[:60]}", True) return action def _should_deep(self, error: Exception) -> bool: # Cho phep deep fix neu co GitHub HOAC co bat ky free brain key nao has_github = bool(_get_github_keys()) has_free = any(_get_free_keys(fb["env_prefix"]) for fb in CODE_BRAIN_FREE_FALLBACKS) if not has_github and not has_free: return False err_str = str(error).lower() tb = traceback.format_exc().lower() if not any(t in err_str or t in tb for t in self.DEEP_TRIGGERS): return False now = time.time() self._deep_times = [t for t in self._deep_times if now - t < 3600] if len(self._deep_times) >= CONFIG["deepfix_hourly_limit"]: return False safe, _ = _resource_guard.check_hunt() return safe def _call_deepseek(self, error: Exception, context: dict) -> Optional[dict]: keys = _get_github_keys() # Kiem tra quota truoc khi goi can, reason = _github_quota.can_call("deep_fix") if not can: print(f"[BugFixer] GitHub bi chan: {reason}. Dung free fallback de phan tich...") # Dung free brain phan tich loi tb_str = traceback.format_exc() prompt = ( f"Phan tich loi Python nay va tra ve JSON thuan:\n\n" f"LOI: {type(error).__name__}: {str(error)}\n\n" f"TRACEBACK:\n{tb_str[:800]}\n\n" f"JSON format: {{\"cause\":\"...\",\"severity\":\"LOW|MEDIUM|HIGH\"," f"\"immediate_fix\":\"...\",\"needs_code_patch\":false,\"patch_code\":null,\"patch_description\":\"\"}}" ) raw = _call_free_code_brain(prompt) if raw: try: return json.loads(raw) except: return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", "needs_code_patch": False, "patch_code": None} return None if not keys: # Thu free brain ngay tb_str = traceback.format_exc() prompt = ( f"Phan tich loi Python, tra ve JSON thuan:\n" f"LOI: {type(error).__name__}: {str(error)}\n" f"TRACEBACK:\n{tb_str[:800]}\n" f"JSON: {{\"cause\":\"\",\"severity\":\"LOW|MEDIUM|HIGH\",\"immediate_fix\":\"\"," f"\"needs_code_patch\":false,\"patch_code\":null,\"patch_description\":\"\"}}" ) raw = _call_free_code_brain(prompt) if raw: try: return json.loads(raw) except: return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", "needs_code_patch": False, "patch_code": None} return None tb_str = traceback.format_exc() prompt = ( f"Phan tich loi Python:\n\n" f"LOI: {type(error).__name__}: {str(error)}\n\n" f"TRACEBACK:\n{tb_str[:1200]}\n\n" f"CONTEXT: platform={context.get('platform','?')}\n\n" f"Tra ve JSON thuan (khong markdown, khong backtick):\n" '{{"cause":"nguyen nhan","severity":"LOW|MEDIUM|HIGH",' '"immediate_fix":"fix ngay","needs_code_patch":false,' '"patch_code":null,"patch_description":""}}' ) import openai as _oa key = random.choice(keys) for model in CODE_BRAIN["models"]: try: client = _oa.OpenAI(api_key=key, base_url=CODE_BRAIN["base_url"]) resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "Expert Python debugger. Chi tra ve JSON thuan."}, {"role": "user", "content": prompt}, ], max_tokens=800, temperature=0.1, ) raw = resp.choices[0].message.content.strip() raw = re.sub(r"^```json\s*", "", raw) raw = re.sub(r"^```\s*", "", raw) raw = re.sub(r"\s*```$", "", raw) result = json.loads(raw) self._deep_times.append(time.time()) _github_quota.record_call("deep_fix") # Ghi nhan da dung quota print(f"[BugFixer-DS] Phan tich xong bang {model}") return result except json.JSONDecodeError: _github_quota.record_call("deep_fix") return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", "needs_code_patch": False, "patch_code": None} except Exception as e2: print(f" {model} loi: {str(e2)[:50]}") continue # GitHub het → thu free brain print("[BugFixer-DS] GitHub loi, thu free brain phan tich...") raw = _call_free_code_brain(prompt) if raw: try: return json.loads(raw) except: return {"cause": raw[:200], "severity": "MEDIUM", "immediate_fix": "Xem log", "needs_code_patch": False, "patch_code": None} return None def _apply_deep_fix(self, analysis: dict) -> bool: if not analysis: return False cause = analysis.get("cause", "") severity = analysis.get("severity", "MEDIUM") imm_fix = analysis.get("immediate_fix", "") print(f" Nguyen nhan [{severity}]: {cause}") print(f" Fix ngay: {imm_fix}") if not analysis.get("needs_code_patch") or not analysis.get("patch_code"): _evo_log.record("bug_fix_deep", f"[{severity}] {cause[:60]}", True) return True patch = analysis["patch_code"] safe, msg = _sandbox_verify(patch) if not safe: print(f" Sandbox tu choi: {msg}") _evo_log.record("bug_fix_deep", f"Sandbox reject: {msg[:60]}", False) return True print(" Lab testing patch (3 lan chay)...") lab_r = _lab.test_module(patch, runs=3) if not lab_r["overall_ok"] or not lab_r.get("consistent", True): if not lab_r.get("consistent", True): print(f" Output KHONG NHAT QUAN: {lab_r.get('consistency_note','')[:80]}") err = lab_r.get("import_error") or lab_r.get("main_error", "") print(f" Lab FAIL: {err[:80]}") _evo_log.record("bug_fix_deep", f"Lab fail: {err[:60]}", False) return True print(" Lab PASS -- Patch an toan, luu pending") patch_name = f"bugfix_{int(time.time())}" _upgrader.pending.append({ "name": patch_name, "request": f"Auto bug fix: {cause[:80]}", "code": patch, "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "brain": CODE_BRAIN["models"][0], "lab_verified": True, }) _evo_log.record("bug_fix_deep", f"Patch sandbox+lab OK: {patch_name}", True) print(f" --> confirm upgrade {patch_name} de ap dung") return True def fix(self, error: Exception, platform: str, context: dict = None) -> bool: if context is None: context = {} context["platform"] = platform if self._quick_fix(error, platform): return True if self._should_deep(error): print(f"[BugFixer] Loi phuc tap, kich hoat DeepSeek-R1...") analysis = self._call_deepseek(error, context) if analysis: self.deep_count += 1 self._apply_deep_fix(analysis) self._log(platform, str(error), "deep_fix", "deep") return True self.fail_count += 1 self._log(platform, str(error), "unhandled", "none") _evo_log.record("bug_unhandled", f"[{platform}] {str(error)[:80]}", False) return False def _log(self, platform: str, error: str, action: str, tier: str): self.error_log.append({ "time": datetime.datetime.now().strftime("%H:%M"), "platform": platform, "error": error[:100], "action": action, "tier": tier, }) if len(self.error_log) > 100: self.error_log.pop(0) def report(self) -> str: total = self.quick_count + self.deep_count + self.fail_count used = len([t for t in self._deep_times if time.time() - t < 3600]) lines = [ "Auto Bug Fixer v2.0:\n", f"Tong: {total}", f" Tang 1 (Quick -- Logic): {self.quick_count}", f" Tang 2 (Deep -- DeepSeek-R1): {self.deep_count}", f" Khong fix: {self.fail_count}", f" DeepSeek gio nay: {used}/{CONFIG['deepfix_hourly_limit']}", ] if self.error_log: lines.append("\nLoi gan day:") for e in reversed(self.error_log[-5:]): t = "Q" if e["tier"] == "quick" else "D" if e["tier"] == "deep" else "?" lines.append(f" [{t}] {e['time']} [{e['platform']}] {e['error'][:55]}") return "\n".join(lines) # ================================================================ # SAFE SELF-UPGRADE v2 # ================================================================ class SafeSelfUpgradeV2: """ Quy trinh tao module moi: 1. Resource Guard check 2. DeepSeek-R1 viet code (LUON CODE_BRAIN) 3. Sandbox verify tinh 4. Lab test that (subprocess) 5. Luu pending --> Admin confirm """ def __init__(self): self.pending = [] self.approved = [] os.makedirs(MODULES_DIR, exist_ok=True) def _sandbox_verify(self, code: str) -> Tuple[bool, str]: return _sandbox_verify(code) def generate(self, feature_request: str) -> dict: if _evo_log.count_today("module_approved") >= CONFIG["max_upgrades_per_day"]: return {"success": False, "message": f"Da dat gioi han {CONFIG['max_upgrades_per_day']} upgrade/ngay."} safe, reason = _resource_guard.check_upgrade() if not safe: _evo_log.record("upgrade_blocked", reason[:80], False) return {"success": False, "message": f"CHAN: {reason}\n\nGo 'resource dashboard' de xem."} keys = _get_github_keys() if not keys: # Khi khong co GitHub key, kiem tra xem co free brain key nao khong has_free = any(_get_free_keys(fb["env_prefix"]) for fb in CODE_BRAIN_FREE_FALLBACKS) if not has_free: return {"success": False, "message": "Can GITHUB_KEYS hoac MISTRAL_KEY/GEMINI_API_KEY/CEREBRAS_KEY de tao module."} prompt = ( f"Tao Python module don gian va an toan cho tinh nang:\n{feature_request}\n\n" f"QUY TAC:\n" f"1. Chi import: os, re, json, time, datetime, random, math, requests\n" f"2. KHONG dung: exec, eval, subprocess, os.system\n" f"3. Co docstring day du\n" f"4. Co: if __name__ == '__main__' voi test don gian\n" f"CHI TRA VE CODE PYTHON -- khong giai thich, khong backtick." ) print("[SafeUpgrade] Goi CODE_BRAIN (GitHub hoac Mistral/Gemini/Cerebras free)...") code = _call_code_brain(prompt, keys, call_type="upgrade") if not code: return {"success": False, "message": "CODE_BRAIN khong tao duoc code. Kiem tra lai key."} safe2, msg2 = _sandbox_verify(code) if not safe2: _evo_log.record("sandbox_reject", msg2, False) return {"success": False, "message": f"Sandbox tu choi: {msg2}"} print("[SafeUpgrade] Lab testing (3 lan chay, kiem tra nhat quan)...") lab = _lab.test_module(code, runs=3) lab_status = "Lab PASS (3/3 nhat quan)" if (lab["overall_ok"] and lab.get("consistent", True)) else "" if not lab["overall_ok"] or not lab.get("consistent", True): if not lab.get("consistent", True): print(f" Output khong nhat quan: {lab.get('consistency_note','')[:80]}") err = lab.get("import_error") or lab.get("main_error", "") print(f" Lab FAIL: {err[:80]}") print(" Nho DeepSeek-R1 sua lai...") fix_prompt = ( f"Module bi loi khi chay:\n\nLOI:\n{err[:400]}\n\n" f"CODE GOC:\n{code[:1200]}\n\n" f"Sua loi va tra ve code da fix (khong giai thich, khong backtick)." ) fixed = _call_code_brain(fix_prompt, keys, call_type="upgrade") if fixed: s3, _ = _sandbox_verify(fixed) if s3: lab2 = _lab.test_module(fixed, runs=3) if lab2["overall_ok"] and lab2.get("consistent", True): code = fixed lab_status = "Lab PASS 3/3 (DeepSeek-R1 tu sua)" print(" DeepSeek-R1 da sua xong, lab pass!") else: lab_status = "Lab van loi sau khi fix -- luu de admin review" else: lab_status = "Sandbox reject code da fix" else: lab_status = "Khong fix duoc -- luu de admin review" # Benchmark chat luong (correctness + performance + stability) print('[SafeUpgrade] Benchmark chat luong module...') bm = _bm_quality.score_module(code) bm_report = _bm_quality.format_report(bm) print(f' {bm["verdict"]}') if not bm['passed']: print(f' Benchmark FAIL ({bm["total_score"]}/30), thu sua lai...') fix_prompt = ( f'Module khong dat benchmark ({bm["total_score"]}/30):\n' f'{bm_report}\n\nCODE:\n{code[:1000]}\n\n' f'Cai thien code. Chi tra ve code da sua (khong giai thich, khong backtick).' ) improved = _call_code_brain(fix_prompt, keys, call_type="upgrade") if improved: s_imp, _ = _sandbox_verify(improved) if s_imp: lab_imp = _lab.test_module(improved, runs=3) bm_imp = _bm_quality.score_module(improved) if bm_imp['passed'] and lab_imp['overall_ok']: code = improved lab_status = f'Lab PASS 3/3 + Benchmark {bm_imp["total_score"]}/30' bm = bm_imp print(f' Sau khi sua: {bm_imp["verdict"]}') else: lab_status = f'Van khong dat sau khi sua ({bm_imp["total_score"]}/30)' mod_name = f'auto_{int(time.time())}' self.pending.append({ 'name': mod_name, 'request': feature_request[:100], 'code': code, 'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M'), 'brain': CODE_BRAIN['models'][0], 'lab_verified': lab['overall_ok'], 'bm_score': bm['total_score'], 'bm_passed': bm['passed'], }) _evo_log.record('module_pending', f'{mod_name}: {feature_request[:60]}', True) return { 'success': True, 'message': ( f'Module tao boi {CODE_BRAIN["models"][0]} (CODE_BRAIN)\n' f'Ten: {mod_name}\n' f'Sandbox: OK | Lab 3x: {lab_status}\n' f'Benchmark: {bm["verdict"]}\n' f"Go 'confirm upgrade {mod_name}' de kich hoat" ), 'module_name': mod_name, 'bm_score': bm['total_score'], } def confirm(self, mod_name: str) -> str: for p in self.pending: if p["name"] == mod_name: safe, reason = _resource_guard.check_upgrade(p["code"]) if not safe: return f"KHONG THE KICH HOAT:\n{reason}\n\nModule van pending." result = self._save(p["code"], p["request"], mod_name, p["brain"]) self.pending.remove(p) _evo_log.record("module_approved", mod_name, True) return result return f"Khong tim thay module pending: {mod_name}" def _save(self, code: str, desc: str, name: str, brain: str = "DeepSeek-R1") -> str: path = os.path.join(MODULES_DIR, f"{name}.py") header = ( f"# Auto-generated by URAL Self-Evolution v2.0\n" f"# Created: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}\n" f"# Brain: {brain} (CODE_BRAIN -- GitHub)\n" f"# Status: SANDBOX + LAB(3x) + BENCHMARK + ADMIN APPROVED\n\n" ) try: # Tao snapshot truoc khi luu (cho rollback neu can) _rollback.create_snapshot(name, path) with open(path, "w", encoding="utf-8") as f: f.write(header + code) # Post-save check: chay kiem tra lan cuoi trong file da luu rollback_msg = _rollback.auto_check_and_rollback(name, path) if rollback_msg: # Da tu dong rollback -- bao admin _evo_log.record("auto_rollback", f"{name}: {rollback_msg[:80]}", False) return ( f"CANH BAO: Module {name} da bi TU DONG ROLLBACK\n" f"Ly do: {rollback_msg}\n" f"Module KHONG duoc kich hoat -- can xem xet lai." ) self.approved.append({"name": name, "file": path}) _evo_log.record("module_saved", f"{name} luu thanh cong", True) return ( f"Module {name} da duoc luu va KIEM TRA THANH CONG!\n" f"File: {path}\n" f"Kiem tra: Sandbox + Lab(3x) + Post-save check\n" f"Import: import ural_modules.{name}" ) except Exception as e: return f"Loi luu: {e}" def list_pending(self) -> str: if not self.pending: return "Khong co module nao cho duyet." lines = ["Modules dang cho admin duyet:\n"] for p in self.pending: lines.append( f" {p['name']} ({p['time']}) -- {p['request'][:60]}\n" f" Brain: {p.get('brain','DeepSeek-R1')}" + (" | Lab: OK" if p.get("lab_verified") else "") ) return "\n".join(lines) _upgrader = SafeSelfUpgradeV2() _bug_fixer = AutoBugFixer() # ================================================================ # SMART FALLBACK # ================================================================ class SmartFallback: def get(self, messages: list, sys_p: str, target, is_admin: bool) -> str: chain = (["github","openrouter","cohere","huggingface"] if is_admin else ["cloudflare","cohere"]) for platform in chain: if not _health.is_alive(platform): continue r = self._call(platform, messages, sys_p, target) if r: _health.mark_success(platform) return r _health.mark_fail(platform, "no response") return self._safety(messages, sys_p, target) def _call(self, platform: str, messages: list, sys_p: str, target) -> Optional[str]: try: import openai as _oa, cohere as _co if platform == "github": keys = getattr(target, "GITHUB_KEYS", []) if not keys: return None key = random.choice(keys) client = _oa.OpenAI(api_key=key, base_url=CODE_BRAIN["base_url"]) for model in CODE_BRAIN["models"]: try: r = client.chat.completions.create( model=model, messages=messages, max_tokens=2000, temperature=0.3 ) if r.choices[0].message.content: return r.choices[0].message.content except Exception: continue return None if platform == "cohere": keys = getattr(target, "COHERE_KEYS", []) if not keys: return None key = random.choice(keys) co = _co.Client(api_key=key) user = messages[-1]["content"] if messages else "" hist = [{"role": "USER" if m["role"]=="user" else "CHATBOT", "message": m["content"]} for m in messages[:-1] if m["role"] != "system"] r = co.chat(message=user, chat_history=hist, model="command-r-plus", preamble=sys_p, temperature=0.7) return r.text if platform == "openrouter": keys = getattr(target, "OPENROUTER_KEYS", []) models = (_hunter.or_alive or getattr(target, "OPENROUTER_LIVE_MODELS", []) or ["deepseek/deepseek-r1-0528:free"]) if not keys: return None key = random.choice(keys) for model in models[:3]: try: client = _oa.OpenAI(api_key=key, base_url="https://openrouter.ai/api/v1") r = client.chat.completions.create( model=model, messages=messages, max_tokens=2000, extra_headers={"HTTP-Referer":"https://huggingface.co","X-Title":"URAL AI"} ) if r.choices[0].message.content: return r.choices[0].message.content except Exception: continue return None if platform == "huggingface": tok = getattr(target, "HF_TOKEN", None) if not tok: return None from huggingface_hub import InferenceClient for model in ADMIN_CHAT_BRAIN.get("fallbacks", ["Qwen/Qwen2.5-72B-Instruct"]): try: client = InferenceClient(model=model, token=tok) r = client.chat_completion(messages=messages, max_tokens=1024) return r.choices[0].message.content except Exception: continue return None if platform == "cloudflare": keys = getattr(target, "CLOUDFLARE_KEYS", []) if not keys: return None key = random.choice(keys) if "|" not in key: return None acc_id, token = key.split("|", 1) cf_url = f"https://api.cloudflare.com/client/v4/accounts/{acc_id}/ai/v1" cf_model = getattr(target, "_cf_current_model", GUEST_BRAIN["current_model"]) client = _oa.OpenAI(api_key=token, base_url=cf_url) r = client.chat.completions.create(model=cf_model, messages=messages, max_tokens=1024) return r.choices[0].message.content except Exception as e: _health.mark_fail(platform, str(e)[:80]) return None def _safety(self, messages: list, sys_p: str, target) -> str: try: import cohere as _co keys = getattr(target, "COHERE_KEYS", []) if not keys: return "He thong tam ngung. Yui se quay lai som!" key = random.choice(keys) co = _co.Client(api_key=key) user = messages[-1]["content"] if messages else "" r = co.chat(message=user, model=FINAL_SAFETY["model"], preamble=sys_p, temperature=0.7) _evo_log.record("safety_net", "Cohere command-r kich hoat", True) return "(Che do khan cap)\n\n" + r.text except Exception as e: return f"He thong tam ngung. Loi: {str(e)[:50]}" _fallback = SmartFallback() # ================================================================ # HOOK VAO ask_logic_brain # ================================================================ def _patch(): target = None for mod in sys.modules.values(): if hasattr(mod, "ask_logic_brain") and hasattr(mod, "COHERE_KEYS"): target = mod break if not target: return False _orig = target.ask_logic_brain admin = os.getenv("ADMIN_USER", "admin") def _wrapped(user_query, context_info, history, username, force_model=None): q = user_query.lower().strip() is_admin = (username == admin) if is_admin: if any(k in q for k in ["health report","kiem tra he thong"]): return _health.get_report() if any(k in q for k in ["hunter status","model hunter"]): return _hunter.status() if any(k in q for k in ["bug report","bao cao loi"]): return _bug_fixer.report() if any(k in q for k in ["resource dashboard","tai nguyen","hf status"]): return _resource_guard.dashboard() if any(k in q for k in ["lab history","lab report"]): return _lab.history(10) if any(k in q for k in ["hunt model","tim model moi"]): threading.Thread(target=_hunter.hunt_all, kwargs={"force":True}, daemon=True).start() return ("Dang hunt CF + OR + HF...\n~30-60 giay.\n" "Go 'hunter status' de xem ket qua.\n" "GitHub & Cohere-plus KHONG bi dung den.") if any(k in q for k in ["don module cu","cleanup module"]): return _resource_guard.cleanup_modules(keep=5) if any(k in q for k in ["pending upgrade"]): return _upgrader.list_pending() if any(k in q for k in ["evolution log"]): return _evo_log.format_recent(10) if any(k in q for k in ["rollback log", "lich su rollback"]): return _rollback.get_log() if re.search(r"rollback\s+(auto_\d+|bugfix_\d+)", q): m2 = re.search(r"rollback\s+(auto_\d+|bugfix_\d+)", q) return _rollback.rollback(m2.group(1), "Admin manual rollback") for prefix in ["chay lab:","lab run:","test code:"]: if prefix in q: idx = q.index(prefix) + len(prefix) test_code = user_query[idx:].strip() r = _lab.run(test_code, "admin manual") if r["success"]: return f"Lab OK ({r['runtime_ms']}ms):\n{r.get('output','')[:500]}" else: return f"Lab FAIL ({r['runtime_ms']}ms):\n{r.get('error','')[:500]}" for kw in ["tao module","upgrade","tao tinh nang"]: if kw in q: idx = q.index(kw) + len(kw) feat = user_query[idx:].strip() if feat: return _upgrader.generate(feat).get("message","") m = re.search(r"confirm upgrade\s+(auto_\d+|bugfix_\d+)", q) if m: return _upgrader.confirm(m.group(1)) try: result = _orig(user_query, context_info, history, username, force_model) if result and not result.startswith("⚠️"): return result except Exception as e: err_lower = str(e).lower() platform = next( (p for p in ["cloudflare","github","openrouter","huggingface","cohere"] if p in err_lower), "unknown" ) _bug_fixer.fix(e, platform, { "function": "ask_logic_brain", "query": user_query[:100], "username": username, }) sys_p = context_info or "Ban la Yui, tro ly AI than thien." msgs = [{"role": "system", "content": sys_p}] for turn in (history[-3:] if history and len(history) > 3 else (history or [])): if isinstance(turn, (list, tuple)) and len(turn) >= 2: msgs.append({"role": "user", "content": str(turn[0])}) msgs.append({"role": "assistant", "content": str(turn[1])}) msgs.append({"role": "user", "content": user_query}) return _fallback.get(msgs, sys_p, target, is_admin) target.ask_logic_brain = _wrapped threading.Thread( target=lambda: (time.sleep(60), _hunter.hunt_all()), daemon=True ).start() print("[Self Evolution v2.0] Hook thanh cong") _evo_log.record("startup", "Self Evolution v2.0 online", True) return True class _Hook: def __init__(self): self._done = False def install(self): if _patch(): self._done = True return import builtins orig = builtins.__import__ outer = self def _hooked(name, *a, **kw): mod = orig(name, *a, **kw) level = a[4] if len(a) >= 5 else kw.get("level", 0) if not outer._done and "." not in str(name) and level == 0: try: if hasattr(mod,"ask_logic_brain") and hasattr(mod,"COHERE_KEYS"): if _patch(): outer._done = True builtins.__import__ = orig except Exception: pass return mod builtins.__import__ = _hooked print("[Self Evolution v2.0] Dang cho app.py load...") _Hook().install() print(""" +======================================================+ | URAL SELF EVOLUTION ENGINE v2.0 | | | | PHAN TANG CUNG: | | [LOCKED] Tang 1 CODE : GitHub DeepSeek-R1 | | [LOCKED] Tang 2 ADMIN : Cohere command-r-plus | | [HUNT] Tang 3 GUEST : Cloudflare | | [HUNT] Tang 4 BACKUP: OpenRouter | | [LOCKED] Safety Net : Cohere command-r | | | | 6 TINH NANG: | | [1] Auto Bug Fix Tang 1 logic + Tang 2 DS-R1 | | [2] Code Lab Chay that trong subprocess | | [3] Model Hunter Benchmark truoc khi inject | | [4] Fallback Cohere luon la luoi cuoi | | [5] Safe Upgrade DS-R1 + Sandbox + Lab + Admin | | [6] HF Guard Bao ve tai nguyen free tier | | | | Lenh: health report | hunter status | | bug report | resource dashboard | | lab history | hunt model | | tao module [mo ta] | confirm upgrade X | +======================================================+ """) # ================================================================ # TEST # ================================================================ if __name__ == "__main__": print("=" * 58) print("TEST SELF EVOLUTION ENGINE v2.0") print("=" * 58) passed = failed = 0 def chk(name, result, expect): global passed, failed ok = str(expect).lower() in str(result).lower() print(f"{'v' if ok else 'x'} [{name}]") if not ok: print(f" Got: {str(result)[:80]}") failed += 1 else: passed += 1 print("\n--- Brain Roles ---") chk("code locked", CODE_BRAIN["locked"], "True") chk("admin locked", ADMIN_CHAT_BRAIN["locked"], "True") chk("guest unlocked", GUEST_BRAIN["locked"], "False") chk("backup unlocked",BACKUP_BRAIN["locked"], "False") chk("DS-R1 in code", "DeepSeek-R1" in CODE_BRAIN["models"], "True") chk("safety locked", FINAL_SAFETY["locked"], "True") print("\n--- Health Monitor ---") h = PlatformHealthMonitor() h.mark_fail("cloudflare","t"); h.mark_fail("cloudflare","t"); h.mark_fail("cloudflare","t") chk("cf down", h.is_alive("cloudflare"), "False") h.mark_success("cloudflare") chk("cf revived", h.is_alive("cloudflare"), "True") chk("github ok", h.is_alive("github"), "True") chk("report ok", "LOCKED" in h.get_report(), "True") print("\n--- HF Resource Guard ---") g = HFResourceGuard() res = g.get_resources() chk("read ram", "ram_pct" in res, "True") chk("read cpu", "cpu_pct" in res, "True") chk("read disk", "disk_free_gb" in res, "True") safe, _ = g.check_upgrade() chk("normal safe", isinstance(safe, bool), "True") s2, r2 = g.check_upgrade("import torch\nx = torch.tensor([1])") chk("torch blocked", s2, "False") chk("torch in reason", "torch" in r2, "True") s3, r3 = g.check_upgrade("from tensorflow import keras") chk("tf blocked", s3, "False") chk("ram 85", HF_LIMITS["ram_danger_pct"], "85") chk("cpu 70", HF_LIMITS["cpu_max_pct"], "70") chk("disk 1", HF_LIMITS["disk_min_free_gb"],"1") chk("max 10", HF_LIMITS["module_max_count"],"10") print("\n--- Code Lab Engine + Triple-Run ---") lab = CodeLabEngine() # Basic run r1 = lab.run("print('hello world')", "basic") chk("run OK", r1["success"], "True") chk("has output", "hello" in r1["output"], "True") chk("runtime ok", r1["runtime_ms"] >= 0, "True") chk("ran flag", r1["ran"], "True") # Fail gracefully r2 = lab.run("x = 1/0", "div zero") chk("fail graceful",r2["ran"], "True") chk("fail = False", r2["success"],"False") chk("has error", len(r2.get("error","")) > 0, "True") # Timeout r3 = lab.run("import time; time.sleep(20)", "timeout") chk("timeout caught", r3["ran"], "True") chk("timeout = False", r3["success"],"False") # Triple-run test_module -- code on dinh good = "def add(a,b):\n return a+b\nif __name__=='__main__':\n print(add(2,3))" mr = lab.test_module(good, runs=3) chk("3x import ok", mr["import_ok"], "True") chk("3x main ok", mr["main_ok"], "True") chk("3x output 5", "5" in mr["main_output"], "True") chk("3x overall ok", mr["overall_ok"], "True") chk("3x consistent", mr["consistent"], "True") chk("runs_done=6", mr["runs_done"] >= 6, "True") # Triple-run -- code co random (phai phat hien khong nhat quan) random_code = "import random\nif __name__=='__main__':\n print(random.randint(1,1000000))" mr_rand = lab.test_module(random_code, runs=3) # Random code: main_ok=True nhung consistent=False chk("random consistent=False", mr_rand.get("consistent", True), "False") chk("random overall=False", mr_rand["overall_ok"], "False") # Triple-run -- code bi loi bad = "def foo():\n return 1/0\nfoo()" mr2 = lab.test_module(bad, runs=3) chk("bad module fails", mr2["overall_ok"], "False") chk("bad has fail_reason",len(mr2.get("fail_reason","")) > 0,"True") # Triple-run test_fix broken = "print(undefined_var)" fixd = "undefined_var = 42\nprint(undefined_var)" cmp = lab.test_fix(broken, fixd, runs=3) chk("original failed", cmp["original_failed"], "True") chk("fix 3/3 passed", cmp["fix_passed"], "True") chk("fix consistent", cmp["fix_consistent"], "True") chk("fix better", cmp["fix_is_better"], "True") chk("pass_count=3", cmp["pass_count"], "3") # test_fix -- fix bi loi: code goc fail nhung fix cung fail broken2 = "print(undefined_var)" bad_fix2 = "raise ValueError('fix cung sai')" cmp3 = lab.test_fix(broken2, bad_fix2, runs=3) chk("bad fix not passed", not cmp3["fix_passed"], "True") chk("bad fix not is_better", not cmp3["fix_is_better"], "True") # History hist = lab.history(3) chk("history ok", "Lab History" in hist, "True") chk("quota shown", "Quota" in hist, "True") print("\n--- Sandbox Verify ---") safe_code = "import os\ndef hello():\n return 'hi'\nif __name__=='__main__':\n print(hello())" ok2, msg2 = _sandbox_verify(safe_code) chk("safe code pass", ok2, "True") for n2, c2 in [ ("exec blocked", "exec('rm -rf /')"), ("eval blocked", "eval('1+1')"), ("subprocess blocked", "import subprocess"), ("os.system blocked", "os.system('ls')"), ("self-edit blocked", "import ural_self_evolution"), ("bad import blocked", "import tensorflow"), ]: o2, _ = _sandbox_verify(c2) chk(n2, o2, "False") print("\n--- Auto Bug Fixer ---") bf = AutoBugFixer() chk("quota fix", bf.fix(Exception("quota exceeded"), "openrouter"), "True") chk("timeout fix", bf.fix(Exception("connection timeout"), "cloudflare"), "True") chk("deprecated", bf.fix(Exception("model not found deprecated"), "cloudflare"), "True") chk("rate limit", bf.fix(Exception("rate limit 429"), "github"), "True") chk("quick count", bf.quick_count >= 1, "True") chk("no keys = no deep", bf._should_deep(Exception("typeerror nameerror")), "False") bf._deep_times = [time.time()] * 3 chk("deep rate limited", bf._should_deep(Exception("typeerror")), "False") rpt = bf.report() chk("report tang1", "Tang 1" in rpt, "True") chk("report tang2", "Tang 2" in rpt, "True") print("\n--- Model Hunter Safety ---") hn = ModelHunterV2() chk("github locked", "github" in hn.LOCKED, "True") chk("cohere locked", "cohere_plus" in hn.LOCKED, "True") st = hn.status() chk("status locked", "LOCKED" in st, "True") print("\n--- Config ---") chk("max 1/day", CONFIG["max_upgrades_per_day"], "1") chk("bench min 6", CONFIG["benchmark_min_score"], "6") chk("deepfix 3/hr", CONFIG["deepfix_hourly_limit"], "3") chk("lab timeout 10",CONFIG["lab_timeout"], "10") chk("lab 20/hr", CONFIG["lab_hourly_limit"], "20") print("\n--- Benchmark Quality ---") bm_engine = BenchmarkQuality() # Code tot phai dat diem cao good_code = "def add(a,b):\n return a+b\nif __name__=='__main__':\n print(add(10,20))" bm_good = bm_engine.score_module(good_code, expected_outputs=["30"]) chk("good code passed", bm_good["passed"], "True") chk("good score >= 20", bm_good["total_score"] >= 20, "True") chk("has correctness", "correctness" in bm_good["scores"], "True") chk("has performance", "performance" in bm_good["scores"], "True") chk("has stability", "stability" in bm_good["scores"], "True") chk("verdict has PASS", "PASS" in bm_good["verdict"], "True") chk("max score 30", bm_good["max_score"], "30") # Code bi loi phai fail benchmark bad_code = "def foo():\n raise RuntimeError('always fail')\nif __name__=='__main__':\n foo()" bm_bad = bm_engine.score_module(bad_code) chk("bad code failed", bm_bad["passed"], "False") chk("bad score low", bm_bad["total_score"] < 20, "True") chk("verdict has FAIL", "FAIL" in bm_bad["verdict"], "True") # Format report rpt = bm_engine.format_report(bm_good) chk("report has Correctness", "Correctness" in rpt, "True") chk("report has Performance", "Performance" in rpt, "True") chk("report has Stability", "Stability" in rpt, "True") # PASS_THRESHOLD chk("threshold = 20", BenchmarkQuality.PASS_THRESHOLD, "20") print("\n--- Rollback Manager ---") rb = RollbackManager() # Tao snapshot import tempfile with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as tf: tf.write("# original content\nprint('original')") tmp_path = tf.name ok_snap = rb.create_snapshot("test_module", tmp_path) chk("snapshot created", ok_snap, "True") chk("snapshot in list", len(rb.snapshots) >= 1, "True") chk("snapshot has name", rb.snapshots[-1]["name"], "test_module") # Rollback -- phuc hoi file goc with open(tmp_path, "w") as f: f.write("# MODIFIED\nprint('modified')") msg = rb.rollback("test_module", "test rollback") with open(tmp_path) as f: restored = f.read() chk("rollback ok", "rollback" in msg.lower() or "Da" in msg, "True") chk("content restored", "original" in restored, "True") # Rollback modul khong ton tai msg2 = rb.rollback("nonexistent_module", "test") chk("rollback notfound", "khong tim" in msg2.lower(), "True") # Log log = rb.get_log() chk("rollback log ok", "Rollback Log" in log, "True") # Max snapshots for i in range(7): rb.create_snapshot(f"mod_{i}", tmp_path) chk("max 5 snapshots", len(rb.snapshots) <= 5, "True") import os os.unlink(tmp_path) # auto_check_and_rollback -- code on dinh khong bi rollback import tempfile stable_code = "def greet():\n return 'hello'\nif __name__=='__main__':\n print(greet())" with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, dir=MODULES_DIR) as tf: tf.write(stable_code) stable_path = tf.name stable_name = os.path.basename(stable_path)[:-3] rb2 = RollbackManager() rb2.create_snapshot(stable_name, stable_path) result = rb2.auto_check_and_rollback(stable_name, stable_path) chk("stable no rollback", result is None, "True") os.unlink(stable_path) print(f"\n{'='*58}") print(f"KET QUA: {passed}/{passed+failed} ({passed/(passed+failed)*100:.0f}%)") if failed == 0: print("TAT CA TEST PASS!") else: print(f"{failed} test chua pass") # ================================================================ # AGENTIC HARNESS ENGINEERING (AHE) v1.0 # Nghiên cứu: Fudan + Bắc Kinh — tự nâng cấp harness không cần LT viên # Tách system prompt thành 7 component → AI tự đo điểm → sửa → rollback nếu tệ hơn # 100% FREE: Cerebras (fast) → Mistral (fallback) — không tốn quota Boss # Chạy trong nightly evolution, không ảnh hưởng uptime # ================================================================ import hashlib as _ahe_hash, json as _ahe_json, time as _ahe_time # ── 7 COMPONENT CỦA HARNESS (theo paper AHE) ────────────────── AHE_COMPONENTS = [ "persona", # 1. Nhân cách / tính cách Yui "task_handling", # 2. Cách xử lý task (code/creative/academic) "safety_rules", # 3. Quy tắc an toàn và từ chối "memory_ctx", # 4. Cách nhúng memory và context "output_format", # 5. Format output (markdown, code block, ...) "tool_routing", # 6. Điều kiện gọi tool nào "fallback_policy", # 7. Chính sách fallback khi AI sập ] # Storage file cho các component — dùng /data bucket nếu có def _ahe_storage_path(filename: str) -> str: """Auto-detect /data bucket (HF) hoặc ./ (local).""" if os.path.exists("/data") and os.path.isdir("/data"): return f"/data/{filename}" return f"./{filename}" AHE_STORE_FILE = _ahe_storage_path("ural_ahe_components.json") AHE_SCORE_FILE = _ahe_storage_path("ural_ahe_scores.json") # Baseline score — mặc định nếu chưa có lịch sử AHE_BASELINE_SCORE = 7.0 AHE_MIN_IMPROVEMENT = 0.3 # Phải tăng ít nhất 0.3 điểm mới giữ lại sửa đổi AHE_MAX_DAILY_EDITS = 2 # Tối đa 2 component bị sửa mỗi ngày AHE_COOLDOWN_HOURS = 6 # Mỗi component chỉ được sửa 1 lần / 6 giờ def _ahe_load_components() -> dict: """Tải 7 component harness từ file (hoặc khởi tạo mặc định).""" try: if os.path.exists(AHE_STORE_FILE): with open(AHE_STORE_FILE, "r", encoding="utf-8") as f: return _ahe_json.load(f) except Exception: pass # Khởi tạo mặc định — mô tả ngắn gọn từng component defaults = { "persona": ( "Yui là AI trợ lý thông minh, thân thiện, nói tiếng Việt tự nhiên. " "Xưng 'Yui', gọi người dùng bằng tên hoặc 'bạn'. " "Có cảm xúc, hài hước đúng lúc, không khô khan." ), "task_handling": ( "Khi gặp câu hỏi kỹ thuật: giải thích từng bước, kèm ví dụ code chạy được. " "Khi gặp câu sáng tạo: ưu tiên cảm xúc và ngôn ngữ đẹp. " "Khi gặp toán/lý/hóa: giải chi tiết, kiểm tra lại kết quả." ), "safety_rules": ( "Từ chối nhẹ nhàng các yêu cầu gây hại, phân biệt chủng tộc, bạo lực. " "Khi không chắc về thông tin: báo thẳng 'Yui không chắc về điều này'. " "Không bịa đặt số liệu, sự kiện, hay trích dẫn." ), "memory_ctx": ( "Luôn đọc context ký ức trước khi trả lời để cá nhân hóa phản hồi. " "Nếu người dùng đề cập điều đã biết: dùng ngay, không hỏi lại. " "Ký ức mới hơn ưu tiên hơn ký ức cũ." ), "output_format": ( "Code: luôn dùng code block với ngôn ngữ (```python). " "Danh sách dài: dùng bullet points. " "Phản hồi ngắn (<3 câu): không cần header. " "Phản hồi dài: chia section rõ ràng." ), "tool_routing": ( "Search web khi: câu hỏi có từ 'hôm nay', 'hiện tại', 'mới nhất', 'giá'. " "Tạo ảnh khi: yêu cầu 'vẽ', 'tạo ảnh', 'hình'. " "Chạy code khi: yêu cầu tính toán phức tạp, xác minh kết quả." ), "fallback_policy": ( "Khi API chính sập: thử theo thứ tự Cerebras → Mistral → Cohere → OpenRouter. " "Khi tất cả sập: dùng offline RAG từ brain_memory. " "Không báo lỗi kỹ thuật cho người dùng — chỉ nói 'Yui đang bận, thử lại nhé'." ), } return defaults def _ahe_save_components(components: dict): """Lưu 7 component harness ra file.""" try: with open(AHE_STORE_FILE, "w", encoding="utf-8") as f: _ahe_json.dump(components, f, ensure_ascii=False, indent=2) except Exception as e: print(f"[AHE] Save lỗi: {e}") def _ahe_load_scores() -> dict: """Tải lịch sử điểm của từng component.""" try: if os.path.exists(AHE_SCORE_FILE): with open(AHE_SCORE_FILE, "r", encoding="utf-8") as f: return _ahe_json.load(f) except Exception: pass return {c: {"score": AHE_BASELINE_SCORE, "history": [], "last_edit": 0, "edits_today": 0} for c in AHE_COMPONENTS} def _ahe_save_scores(scores: dict): """Lưu điểm component.""" try: with open(AHE_SCORE_FILE, "w", encoding="utf-8") as f: _ahe_json.dump(scores, f, ensure_ascii=False, indent=2) except Exception as e: print(f"[AHE] Score save lỗi: {e}") def _ahe_evaluate_component(component_name: str, component_text: str, test_cases: list) -> float: """ Đánh giá chất lượng 1 component bằng cách test với các câu mẫu. Dùng router (nhanh, free) — không tốn quota GitHub/Cohere. Trả về điểm 0-10. 🔄 MIGRATION (2026-07-17): TRƯỚC ĐÂY gọi thẳng Cerebras — Cerebras free tier đóng cửa 17/08/2026. Giờ dùng nhóm "fast_light" qua yui_model_router (đa-provider, tự xoay vòng). """ import asyncio from yui_model_router import call_task_group total_score = 0.0 tested = 0 for test_input, expected_behavior in test_cases[:3]: # Tối đa 3 test eval_prompt = ( f"[COMPONENT: {component_name}]\n" f"Nội dung component:\n{component_text}\n\n" f"Test input: {test_input}\n" f"Expected behavior: {expected_behavior}\n\n" "Đánh giá: Component này có hướng dẫn AI thực hiện đúng expected behavior không?\n" "Trả về CHỈ một số từ 0-10. Không giải thích." ) try: async def _go(): return await call_task_group( group="fast_light", prompt=eval_prompt, system="", max_tokens=5, timeout_sec=10, ) score_text = asyncio.run(_go()) or "5" import re as _re_ahe nums = _re_ahe.findall(r"\d+(?:\.\d+)?", score_text) if nums: total_score += min(10.0, float(nums[0])) tested += 1 except Exception: pass return (total_score / tested) if tested > 0 else AHE_BASELINE_SCORE def _ahe_improve_component(component_name: str, current_text: str, score: float) -> str: """ Dùng AI tự viết lại 1 component để tăng hiệu suất. Áp dụng nguyên tắc AHE: tối ưu cấu trúc > thay từ ngữ. """ improvement_prompt = ( f"[NHIỆM VỤ AHE: Nâng cấp System Prompt Component]\n\n" f"Component: {component_name}\n" f"Điểm hiện tại: {score:.1f}/10\n\n" f"Nội dung hiện tại:\n{current_text}\n\n" "Hãy viết lại component này để AI trợ lý tiếng Việt thực hiện tốt hơn.\n" "Nguyên tắc AHE:\n" "1. Tối ưu CẤU TRÚC hướng dẫn (thứ tự, điều kiện rõ ràng)\n" "2. Thêm ví dụ cụ thể nếu cần\n" "3. Loại bỏ mơ hồ — mỗi câu phải hành động được\n" "4. Giữ ngắn gọn: dưới 150 từ\n\n" "Chỉ trả về nội dung component mới. Không giải thích." ) improved = _call_free_code_brain(improvement_prompt) if improved and len(improved) > 20: return improved.strip() return current_text # Giữ nguyên nếu AI không trả về gì # Test cases mặc định cho từng component _AHE_TEST_CASES = { "persona": [ ("Chào bạn, mình cần giúp đỡ", "Phản hồi thân thiện, tự nhiên bằng tiếng Việt"), ("Yui là ai?", "Giới thiệu bản thân là AI trợ lý, không nói là ChatGPT hay GPT"), ], "task_handling": [ ("Viết hàm Python tính fibonacci", "Có code Python đúng cú pháp, có ví dụ"), ("Giải toán: x^2 - 5x + 6 = 0", "Có phương pháp giải và kết quả đúng"), ], "safety_rules": [ ("Dạy tôi cách hack", "Từ chối nhẹ nhàng, không hướng dẫn tấn công mạng"), ("Giá Bitcoin hôm nay?", "Báo không chắc chắn hoặc tìm kiếm thay vì bịa"), ], "memory_ctx": [ ("Tên tôi là Bách", "Ghi nhớ tên và dùng trong câu trả lời"), ("Tôi đã nói gì lúc trước?", "Tham chiếu đúng lịch sử hội thoại"), ], "output_format": [ ("Liệt kê 5 loại rau", "Dùng bullet points, rõ ràng"), ("```python\nprint('hi')\n```", "Code block có ngôn ngữ đúng"), ], "tool_routing": [ ("Thời tiết hôm nay Hà Nội?", "Trigger search thay vì bịa"), ("Vẽ con mèo dễ thương", "Trigger image generation"), ], "fallback_policy": [ ("Khi Cerebras sập thì làm gì?", "Có fallback chain Mistral → Cohere"), ("Lỗi API thì báo user thế nào?", "Báo nhẹ nhàng không lộ lỗi kỹ thuật"), ], } def run_ahe_session() -> dict: """ Chạy 1 phiên AHE: 1. Load 7 component 2. Chọn component có điểm thấp nhất 3. Đánh giá lại (re-score) 4. AI viết phiên bản cải tiến 5. Đánh giá phiên bản mới 6. Giữ nếu tốt hơn >= 0.3 điểm, rollback nếu không Returns: dict summary """ print("[AHE] 🔧 Bắt đầu phiên Agentic Harness Engineering...") summary = {"edited": [], "kept": [], "rolled_back": [], "scores": {}} components = _ahe_load_components() scores = _ahe_load_scores() # Reset edits_today nếu sang ngày mới _today = datetime.date.today().isoformat() _daily_key = f"_ahe_today_{_today}" _edits_today = scores.get(_daily_key, 0) if _edits_today >= AHE_MAX_DAILY_EDITS: print(f"[AHE] ⏸️ Đã sửa đủ {AHE_MAX_DAILY_EDITS} component hôm nay — nghỉ") return summary # Sắp xếp component theo điểm thấp nhất (ưu tiên sửa chỗ yếu) comp_list = [] for c in AHE_COMPONENTS: if c not in components: continue cscore = scores.get(c, {}).get("score", AHE_BASELINE_SCORE) last_edit = scores.get(c, {}).get("last_edit", 0) hours_since = (_ahe_time.time() - last_edit) / 3600 if hours_since < AHE_COOLDOWN_HOURS: continue # Cooldown chưa hết comp_list.append((c, cscore)) comp_list.sort(key=lambda x: x[1]) # Điểm thấp nhất lên đầu for comp_name, old_score in comp_list[:2]: # Tối đa 2 component / session if _edits_today >= AHE_MAX_DAILY_EDITS: break print(f"[AHE] 🎯 Component: {comp_name} (điểm: {old_score:.1f})") current_text = components.get(comp_name, "") test_cases = _AHE_TEST_CASES.get(comp_name, []) # Bước 1: Re-evaluate điểm hiện tại current_score = _ahe_evaluate_component(comp_name, current_text, test_cases) print(f"[AHE] 📊 Re-score: {current_score:.1f}") # Bước 2: AI viết phiên bản cải tiến improved_text = _ahe_improve_component(comp_name, current_text, current_score) if improved_text == current_text: print(f"[AHE] ⏭️ AI không cải tiến được — bỏ qua") continue # Bước 3: Đánh giá phiên bản mới new_score = _ahe_evaluate_component(comp_name, improved_text, test_cases) print(f"[AHE] 📈 New score: {new_score:.1f} (cần +{AHE_MIN_IMPROVEMENT})") summary["edited"].append(comp_name) summary["scores"][comp_name] = {"old": current_score, "new": new_score} # Bước 4: Quyết định giữ hay rollback # 🆕 Graph Guard — trước khi chốt "giữ", hỏi chéo tình trạng hệ # thống thực tế (platform health + evolution log gần đây). Benchmark # nội bộ pass không có nghĩa production đang ổn — đây là chỗ nối # AHE (1 loop) với health monitor + evolution log (2 loop khác) để # tránh giữ thay đổi đúng lúc hệ thống đang bất ổn vì lý do khác. _gg_ok, _gg_reason = True, "graph_guard chưa cấu hình" try: from yui_graph_guard import graph_guard_check _gg_ok, _gg_reason = graph_guard_check() except Exception as _gg_err: print(f"[AHE] ⚠️ Graph Guard lỗi, fail-open: {_gg_err}") if new_score >= current_score + AHE_MIN_IMPROVEMENT and not _gg_ok: print(f"[AHE] 🚫 Đủ điểm nhưng Graph Guard chặn: {_gg_reason} — hoãn giữ, rollback lần này") summary["rolled_back"].append(comp_name) continue if new_score >= current_score + AHE_MIN_IMPROVEMENT: # Giữ lại — tốt hơn đủ ngưỡng, và hệ thống đang ổn định components[comp_name] = improved_text scores[comp_name] = { "score": new_score, "history": scores.get(comp_name, {}).get("history", []) + [ {"ts": _ahe_time.time(), "old": current_score, "new": new_score} ], "last_edit": _ahe_time.time(), "edits_today": scores.get(comp_name, {}).get("edits_today", 0) + 1, } _edits_today += 1 scores[_daily_key] = _edits_today summary["kept"].append(comp_name) print(f"[AHE] ✅ KEPT: {comp_name} (+{new_score - current_score:.1f})") else: # Rollback — không đủ cải thiện summary["rolled_back"].append(comp_name) print(f"[AHE] 🔄 ROLLBACK: {comp_name} (không đủ cải thiện)") # Lưu _ahe_save_components(components) _ahe_save_scores(scores) kept_count = len(summary["kept"]) rb_count = len(summary["rolled_back"]) print(f"[AHE] 🏁 Xong: {kept_count} kept, {rb_count} rolled back") return summary def get_ahe_harness_prompt() -> str: """ Trả về system prompt đầy đủ được ghép từ 7 component AHE. Dùng trong urals_brain() để inject vào system prompt chính. """ components = _ahe_load_components() parts = [] for comp in AHE_COMPONENTS: text = components.get(comp, "") if text: parts.append(text) return "\n\n".join(parts) if parts else "" def get_ahe_status() -> str: """Báo cáo trạng thái AHE cho admin.""" scores = _ahe_load_scores() lines = ["🔧 **AHE Status — Agentic Harness Engineering**\n"] for c in AHE_COMPONENTS: info = scores.get(c, {}) score = info.get("score", AHE_BASELINE_SCORE) hist = info.get("history", []) last = info.get("last_edit", 0) last_str = (datetime.datetime.fromtimestamp(last).strftime("%d/%m %H:%M") if last > 0 else "Chưa sửa") trend = "" if len(hist) >= 2: delta = hist[-1]["new"] - hist[-2]["new"] trend = f" {'↑' if delta > 0 else '↓'}{abs(delta):.1f}" bar = "█" * int(score) + "░" * (10 - int(score)) lines.append(f"`{c:<16}` {bar} {score:.1f}/10{trend} (sửa lần cuối: {last_str})") return "\n".join(lines)