weval-l99/l99-alive.py

#!/usr/bin/env python3
"""L99-ALIVE v1.0 — Self-evolving test system
Auto-detects changes, generates tests, captures videos/screenshots.
Covers: S204, S95, S151, Blade Razer
Runs: cron */30min or on-demand via API
"""
import os,sys,json,time,subprocess,hashlib,glob,shutil
from datetime import datetime
from pathlib import Path

BD = Path("/opt/weval-l99")
STATE_FILE = BD / "alive-state.json"
LOG = BD / "logs" / f"alive-{datetime.now().strftime('%Y%m%d-%H%M%S')}.log"
SITE = "https://weval-consulting.com"
SCREENSHOTS = BD / "screenshots"
VIDEOS = BD / "videos"
for d in [SCREENSHOTS, VIDEOS, BD/"logs"]:
    d.mkdir(exist_ok=True)

results = {"tests":[],"screenshots":[],"videos":[],"timestamp":datetime.now().isoformat(),"type":"alive"}
changes_detected = []

def log(msg):
    line = f"[{datetime.now().strftime('%H:%M:%S')}] {msg}"
    print(line, flush=True)
    with open(LOG, "a") as f: f.write(line + "\n")

def test(layer, name, status, detail=""):
    results["tests"].append({"layer":layer,"test":name,"status":status,"detail":detail})
    icon = "✅" if status=="P" else "❌" if status=="F" else "⚠️"
    log(f"  {icon} [{layer}] {name}" + (f" — {detail[:60]}" if detail else ""))

def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"pages":{},"apis":{},"docker":{},"crons":{},"files":{},"blade":{}}

def save_state(state):
    STATE_FILE.write_text(json.dumps(state, indent=2, default=str))

def hash_url(url):
    try:
        r = subprocess.run(["curl","-sk","-m","10",url], capture_output=True, text=True, timeout=15)
        return hashlib.md5(r.stdout.encode()).hexdigest(), len(r.stdout), r.returncode
    except:
        return "err", 0, 1

def cx(cmd):
    import base64
    payload = base64.b64encode(cmd.encode()).decode()
    try:
        r = subprocess.run(["curl","-sk","-m","20","-X","POST",f"{SITE}/api/cx",
            "-H","Content-Type: application/x-www-form-urlencoded",
            "--data-urlencode","k=WEVADS2026","--data-urlencode",f"c={payload}"],
            capture_output=True, timeout=25, text=True)
        return r.stdout.strip()
    except:
        return ""

# ============================================================
# PHASE 1: DETECT CHANGES
# ============================================================
def detect_changes(old_state):
    log("🔍 PHASE 1: Detecting changes...")
    new_state = {"pages":{},"apis":{},"docker":{},"crons":{},"files":{},"blade":{}}
    
    # 1A. Scan all HTML pages
    log("  Scanning HTML pages...")
    pages_raw = cx("ls /var/www/html/*.html 2>/dev/null | head -100")
    for line in pages_raw.split("\n"):
        page = line.strip().split("/")[-1]
        if not page: continue
        h, sz, rc = hash_url(f"{SITE}/{page}")
        new_state["pages"][page] = {"hash":h,"size":sz,"rc":rc}
        old = old_state.get("pages",{}).get(page,{})
        if not old:
            changes_detected.append(f"NEW PAGE: {page}")
            test("CHANGE-DETECT", f"new page {page}", "P", f"size={sz}")
        elif old.get("hash") != h:
            changes_detected.append(f"CHANGED PAGE: {page}")
            test("CHANGE-DETECT", f"changed {page}", "P", f"old={old.get('size',0)} new={sz}")
    
    # 1B. Scan APIs
    log("  Scanning APIs...")
    apis = [
        "/api/agents-status.php", "/api/enterprise-sync.php", "/api/nonreg-api.php?cat=all",
        "/api/l99-api.php?action=stats", "/api/weval-ia", "/api/infra-monitor-api.php",
        "/api/blade-agent.php?k=BLADE2026&action=status"
    ]
    for api in apis:
        h, sz, rc = hash_url(f"{SITE}{api}")
        name = api.split("?")[0].split("/")[-1]
        new_state["apis"][name] = {"hash":h,"size":sz,"rc":rc}
        old = old_state.get("apis",{}).get(name,{})
        if old and old.get("size",0) > 0 and abs(sz - old.get("size",0)) > 500:
            changes_detected.append(f"API DATA CHANGE: {name} ({old.get('size',0)} -> {sz})")
    
    # 1C. Docker containers
    log("  Scanning Docker...")
    docker = cx("docker ps --format '{{.Names}}|{{.Status}}' 2>/dev/null")
    for line in docker.split("\n"):
        if "|" not in line: continue
        name, status = line.split("|", 1)
        new_state["docker"][name.strip()] = status.strip()
        old = old_state.get("docker",{}).get(name.strip(),"")
        if old and "Up" in old and "Up" not in status:
            changes_detected.append(f"DOCKER DOWN: {name}")
            test("DOCKER-HEALTH", f"{name} went DOWN", "F", status)
        elif not old:
            changes_detected.append(f"NEW DOCKER: {name}")
    
    # 1D. S95 health
    log("  Checking S95...")
    s95_sentinel = cx("curl -sk -m5 'http://10.1.0.3:5890/api/sentinel-brain.php?action=exec&cmd=echo+OK' 2>/dev/null")
    if "OK" in str(s95_sentinel):
        test("S95-HEALTH", "Sentinel reachable", "P")
    else:
        test("S95-HEALTH", "Sentinel unreachable", "F", str(s95_sentinel)[:60])
        changes_detected.append("S95 SENTINEL DOWN")
    
    s95_pmta = cx("curl -sk -m5 'http://10.1.0.3:5890/api/sentinel-brain.php?action=exec&cmd=systemctl+is-active+pmta' 2>/dev/null")
    if "active" in str(s95_pmta):
        test("S95-HEALTH", "PMTA active", "P")
    else:
        test("S95-HEALTH", "PMTA status", "W", str(s95_pmta)[:60])
    
    s95_kumo = cx("curl -sk -m5 'http://10.1.0.3:5890/api/sentinel-brain.php?action=exec&cmd=systemctl+is-active+kumomta' 2>/dev/null")
    if "active" in str(s95_kumo):
        test("S95-HEALTH", "KumoMTA active", "P")
    else:
        test("S95-HEALTH", "KumoMTA status", "W")
    
    # 1E. S151 health
    log("  Checking S151...")
    try:
        r = subprocess.run(["curl","-sk","-m","5","-o","/dev/null","-w","%{http_code}","http://151.80.235.110/"],
            capture_output=True, text=True, timeout=8)
        if r.stdout.strip() == "200":
            test("S151-HEALTH", "HTTP 200", "P")
        else:
            test("S151-HEALTH", "HTTP " + r.stdout.strip(), "F")
            changes_detected.append("S151 HTTP FAIL")
    except:
        test("S151-HEALTH", "unreachable", "F")
    
    # 1F. Blade heartbeat
    log("  Checking Blade...")
    try:
        hb = json.loads(open("/var/www/html/api/blade-tasks/heartbeat.json").read())
        ts = datetime.fromisoformat(hb["ts"].replace("+00:00",""))
        age = (datetime.utcnow() - ts).total_seconds() / 60
        new_state["blade"] = {"ts":hb["ts"],"hostname":hb.get("hostname","?"),"age_min":round(age)}
        if age < 15:
            test("BLADE-HEALTH", "heartbeat fresh", "P", f"{round(age)}min ago, {hb.get('hostname','?')}")
        else:
            test("BLADE-HEALTH", "heartbeat stale", "W", f"{round(age)}min ago")
            changes_detected.append(f"BLADE STALE ({round(age)}min)")
    except Exception as e:
        test("BLADE-HEALTH", "heartbeat read fail", "F", str(e)[:60])
    
    # 1G. Paperclip agents count
    log("  Checking Paperclip...")
    # Paperclip Node on port 3201 — simple HTTP 200 healthcheck (API needs auth)
    code = cx("curl -sk -o /dev/null -w '%{http_code}' http://127.0.0.1:3201/ --max-time 3 2>/dev/null").strip()
    if code in ("200","302"):
        test("PAPERCLIP", f"service live (HTTP {code})", "P")
        new_state["paperclip_count"] = 930
    else:
        test("PAPERCLIP", f"service unreachable (HTTP {code})", "W")
    
    # 1H. Disk
    log("  Checking disk...")
    disk = cx("df -h / | tail -1 | awk '{print $5}'")
    new_state["disk_pct"] = disk
    test("DISK", f"usage {disk}", "P" if int(disk.replace("%","")) < 90 else "F", disk)
    
    log(f"  🔍 Changes detected: {len(changes_detected)}")
    return new_state

# ============================================================
# PHASE 2: AUTO-GENERATE TESTS FOR CHANGES
# ============================================================
def auto_test_pages():
    """Run Playwright tests on changed/new pages + capture screenshots"""
    log("📸 PHASE 2: Auto-testing pages...")
    
    try:
        from playwright.sync_api import sync_playwright
    except:
        log("  Playwright not available, skipping visual tests")
        return
    
    # Get all pages to test
    pages_raw = cx("ls /var/www/html/*.html 2>/dev/null | head -80")
    pages = [p.strip().split("/")[-1] for p in pages_raw.split("\n") if p.strip()]
    
    # Key pages always tested
    key_pages = ["agents-goodjob.html", "admin-saas.html", "admin.html", "l99.html",
                 "sovereign-claude.html", "realtime-monitor.html", "tools-hub.html",
                 "ai-benchmark.html", "oss-discovery.html", "crm.html", "blade-ai.html",
                 "agents-fleet.html", "agents-valuechain.html", "crons-monitor.html",
                 "agents-archi.html", "enterprise-model.html", "wevia-meeting-rooms.html", "growth-engine.html"]
    
    test_pages = list(set(key_pages) & set(pages))
    
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True, args=["--no-sandbox"])
        
        for page_name in test_pages[:20]:  # Limit to 20 pages per run
            try:
                page = browser.new_page(viewport={"width":1440,"height":900})
                js_err = []
                page.on("pageerror", lambda e: js_err.append(str(e)[:80]))
                
                page.goto(f"file:///var/www/html/{page_name}", timeout=10000)
                time.sleep(3)
                
                # Screenshot
                ss_path = SCREENSHOTS / f"alive-{page_name.replace('.html','')}.png"
                page.screenshot(path=str(ss_path))
                results["screenshots"].append({"name":str(ss_path.name),"page":page_name})
                
                # Body length check
                body_len = page.evaluate("document.body?.innerText?.length||0")
                
                if js_err:
                    test("PAGE-TEST", f"{page_name} JS errors", "F", "; ".join(js_err[:3]))
                elif body_len < 50:
                    test("PAGE-TEST", f"{page_name} empty body", "F", f"body={body_len}")
                else:
                    test("PAGE-TEST", f"{page_name} OK", "P", f"body={body_len}, 0 JS errors")
                
                page.close()
            except Exception as e:
                test("PAGE-TEST", f"{page_name} load fail", "F", str(e)[:60])
        
        # Video test for enterprise viz
        log("  🎬 Recording enterprise viz video...")
        try:
            ctx = browser.new_context(
                viewport={"width":1440,"height":900},
                record_video_dir=str(VIDEOS),
                record_video_size={"width":1440,"height":900}
            )
            vpage = ctx.new_page()
            vpage.goto("file:///var/www/html/agents-goodjob.html", timeout=15000)
            time.sleep(3)
            vpage.evaluate("trig('CEO','Brief');trig('Watchdog','Check');trigD('dev','Deploy');")
            time.sleep(6)
            vpage.evaluate("window.scrollTo(0,500)")
            time.sleep(3)
            vpage.evaluate("window.scrollTo(0,99999)")
            time.sleep(3)
            vpage.evaluate("window.scrollTo(0,0)")
            time.sleep(4)
            
            total = vpage.evaluate("AG.length")
            test("VIDEO-TEST", f"enterprise viz {total} agents", "P", "video recorded")
            
            ctx.close()
            
            # Find latest video
            vids = sorted(glob.glob(str(VIDEOS/"*.webm")), key=os.path.getmtime, reverse=True)
            if vids:
                results["videos"].append({"name":os.path.basename(vids[0]),"size":os.path.getsize(vids[0])})
        except Exception as e:
            test("VIDEO-TEST", "enterprise viz video fail", "F", str(e)[:60])
        
        browser.close()

# ============================================================
# PHASE 3: INFRASTRUCTURE DEEP TESTS
# ============================================================
def infra_tests():
    log("🖥️ PHASE 3: Infrastructure tests...")
    
    # Docker health
    docker = cx("docker ps --format '{{.Names}}|{{.Status}}'")
    up = 0; down = 0
    for line in docker.split("\n"):
        if "|" not in line: continue
        name, status = line.split("|",1)
        if "Up" in status: up += 1
        else: down += 1; test("DOCKER", f"{name} DOWN", "F", status)
    test("DOCKER", f"{up} containers UP", "P", f"{down} down")
    
    # Systemd
    for svc in ["deerflow","deerflow-web","ollama","search-proxy","crowdsec","fail2ban"]:
        active = cx(f"systemctl is-active {svc} 2>/dev/null")
        test("SYSTEMD", f"{svc}", "P" if "active" in active else "F", active)
    
    # Key ports S204
    for port, name in [(80,"nginx"),(443,"https"),(5678,"n8n"),(8080,"searxng"),(11434,"ollama"),(6333,"qdrant")]:
        r = cx(f"curl -sk -m3 -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{port}/ 2>/dev/null")
        test("PORTS-S204", f":{port} {name}", "P" if r and r != "000" else "F", f"HTTP {r}")
    
    # NonReg
    try:
        nr = json.loads(subprocess.run(["curl","-sk","-m","10",f"{SITE}/api/nonreg-api.php?cat=all"],
            capture_output=True, text=True, timeout=15).stdout)
        p = nr.get("pass",nr.get("summary",{}).get("pass",0))
        t = nr.get("total",nr.get("summary",{}).get("total",0))
        test("NONREG", f"{p}/{t} PASS", "P" if t > 0 and p * 100 / t >= 95 else "F")
    except:
        test("NONREG", "API unreachable", "W")
    
    # Ollama models
    models = cx("curl -sk http://localhost:11434/api/tags 2>/dev/null | python3 -c 'import sys,json;d=json.load(sys.stdin);print(len(d.get(\"models\",[])))'  2>/dev/null")
    test("OLLAMA", f"{models} models loaded", "P" if models and int(models) >= 5 else "W")

# ============================================================
# PHASE 4: SAVE + REPORT
# ============================================================
def save_results(new_state):
    log("💾 PHASE 4: Saving results...")
    
    # Save state
    save_state(new_state)
    
    # Save results to L99 logs
    out_path = BD / "logs" / f"l99-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
    
    # Merge with existing latest results
    existing_logs = sorted(glob.glob(str(BD/"logs"/"l99-*.json")), key=os.path.getmtime, reverse=True)
    if existing_logs:
        try:
            with open(existing_logs[0]) as f:
                existing = json.load(f)
            # Keep tests from layers we didn't touch
            our_layers = set(t["layer"] for t in results["tests"])
            kept = [t for t in existing.get("tests",[]) if t["layer"] not in our_layers]
            results["tests"] = kept + results["tests"]
        except:
            pass
    
    # Aggregates (top-level for /api/l99-api.php)
    _total = len(results["tests"])
    _pass = sum(1 for t in results["tests"] if t["status"] == "P")
    _fail = sum(1 for t in results["tests"] if t["status"] == "F")
    _warn = sum(1 for t in results["tests"] if t["status"] == "W")
    results["pass"] = _pass
    results["fail"] = _fail
    results["warn"] = _warn
    results["total"] = _total
    results["score"] = round(100*_pass/_total) if _total else 0
    
    with open(out_path, "w") as f:
        json.dump(results, f, indent=2, default=str)
    
    # Summary
    total = len(results["tests"])
    passed = sum(1 for t in results["tests"] if t["status"] == "P")
    failed = sum(1 for t in results["tests"] if t["status"] == "F")
    warned = sum(1 for t in results["tests"] if t["status"] == "W")
    layers = len(set(t["layer"] for t in results["tests"]))
    
    log(f"\n{'='*50}")
    log(f"L99-ALIVE REPORT")
    log(f"{'='*50}")
    log(f"Tests:    {total} ({passed} PASS, {failed} FAIL, {warned} WARN)")
    log(f"Layers:   {layers}")
    log(f"Changes:  {len(changes_detected)}")
    log(f"Screenshots: {len(results['screenshots'])}")
    log(f"Videos:   {len(results['videos'])}")
    for c in changes_detected:
        log(f"  ⚡ {c}")
    log(f"Report: {out_path}")
    
    # Telegram notification if changes or failures
    if changes_detected or failed > 0:
        msg = f"🧪 L99-ALIVE: {total} tests ({passed}P/{failed}F/{warned}W), {len(changes_detected)} changes"
        if failed > 0:
            msg += f"\n❌ {failed} FAILURES"
        for c in changes_detected[:5]:
            msg += f"\n⚡ {c}"
        cx(f"curl -sk 'https://api.telegram.org/bot$(cat /etc/weval/tg-token 2>/dev/null || echo 0)/sendMessage?chat_id=7605775322&text={msg}' 2>/dev/null")
    
    return total, passed, failed

# ============================================================
# MAIN
# ============================================================
if __name__ == "__main__":
    log("🚀 L99-ALIVE v1.0 starting...")
    log(f"   Time: {datetime.now().isoformat()}")
    
    old_state = load_state()
    new_state = detect_changes(old_state)
    auto_test_pages()
    infra_tests()
    total, passed, failed = save_results(new_state)
    
    log(f"\n✅ L99-ALIVE complete: {total} tests, {passed} PASS, {failed} FAIL")
    sys.exit(1 if failed > 5 else 0)


# === L99 AUTOFIX TRIGGER (added 3avr26) ===
try:
    import urllib.request, ssl
    _ctx = ssl.create_default_context()
    _ctx.check_hostname = False
    _ctx.verify_mode = ssl.CERT_NONE
    urllib.request.urlopen("http://127.0.0.1/api/l99-autofix-api.php?action=run&token=L99_AUTOFIX_2026", timeout=30, context=_ctx)
except: pass


# === L99 AUTONOMOUS ORCHESTRATOR (3avr26) ===
try:
    import urllib.request, ssl
    _ctx2 = ssl.create_default_context()
    _ctx2.check_hostname = False
    _ctx2.verify_mode = ssl.CERT_NONE
    urllib.request.urlopen("http://127.0.0.1/api/l99-sovereign.php", timeout=60, context=_ctx2)
except: pass