app.py

"""
THE Z AI — Computer Mode Server v20 "Headless-Native"
=====================================================================================
إعادة كتابة كاملة من الصفر — بنية مختلفة تماماً عن v11-v13.

── لماذا إعادة كتابة كاملة؟ ──
البنية القديمة (v11-v13) اعتمدت على:
   Xvfb (X server وهمي منفصل) + Firefox/Chromium عادي يرسم عليه + التقاط
   الشاشة بأداة خارجية (import/ImageMagick، أو python-xlib، أو xwd) + xdotool
   لكل حركة فأرة/كيبورد.
هذا يعني *عمليتين منفصلتين* لكل مستخدم (Xvfb + المتصفح) + عملية subprocess
ثالثة لكل لقطة شاشة واحدة، وكلها تتواصل عبر بروتوكول X11 الثقيل. على سيرفر
موارد محدودة (HF Free: 2 vCPU/16GB مشتركة)، هذا يعني استهلاك ذاكرة/CPU أعلى
بكثير من اللازم، وأي بطء عابر في X11 يُفسَّر كـ"فشل التقاط" فيُفعّل ريستارت
تلقائي للجلسة كاملة — وهذا بالضبط ما كان يسبب الريستارت المتكرر والأخطاء.

── البنية الجديدة v20 ──
   Chromium بوضع "--headless=new" (الوضع الرسمي الحديث من Google) *فقط*.
   بدون Xvfb. بدون X server من أي نوع. Chromium يرندر الصفحة بالكامل داخلياً
   في الذاكرة (نفس محرك الرندر الحقيقي، بدون نافذة مرئية)، ونتحدث معه حصرياً
   عبر Chrome DevTools Protocol (CDP) عبر WebSocket واحد لكل مستخدم:
     - Page.captureScreenshot   → لقطة الشاشة (JPEG) مباشرة من المتصفح نفسه،
                                   بدون subprocess خارجي، بدون فحص "هل الصورة
                                   سوداء" (headless لا يُنتج شاشات سوداء أصلاً
                                   لأنه لا يعتمد على X server قد يتعطل).
     - Input.dispatchMouseEvent → حركة/نقر/سحب الفأرة (بدل xdotool mousemove/click)
     - Input.dispatchKeyEvent   → كتابة/اختصارات لوحة المفاتيح (بدل xdotool key/type)
     - Page.navigate            → فتح رابط/تبويب جديد
     - Runtime.evaluate         → لصق النص العربي (clipboard بديل)، تنفيذ سكربت مساعد

   نتيجة هذا التبسيط:
     - عملية واحدة فقط لكل مستخدم (Chromium) بدل عمليتين (Xvfb+متصفح).
     - استهلاك ذاكرة أقل بشكل ملحوظ (لا X server منفصل يُبقي framebuffer في الذاكرة).
     - لا وجود لمفهوم "فشل التقاط الشاشة" الذي كان يُفعّل ريستارت — لو فشل CDP
       فالمشكلة الوحيدة الممكنة هي أن Chromium نفسه توقف، فنُعيد تشغيله هو فقط
       (لا حاجة لأي "reset" أوسع يمسح بروفايلات أو Xvfb لأنه غير موجود أصلاً).
     - أخف تعقيداً: أداة تحكم واحدة (CDP) بدل خليط (xdotool + import + xlib + xwd).

── ما الذي أُزيل من v13 ولماذا ──
   أدوات "browser_get_elements / browser_click_element / browser_type_into_element /
   browser_find_element / browser_get_page_text / browser_execute_js / browser_navigate
   (CDP) / browser_scroll_to_element / browser_get_state" — هذه لم تكن تُستخدم إطلاقاً
   من الواجهة الأمامية (index.html) بعد فحص الكود الفعلي؛ الواجهة تُرسل فقط:
   screenshot, terminal, mouse_move, mouse_click, mouse_drag, keyboard_type,
   keyboard_hotkey, keyboard_press, scroll, clipboard_write, clipboard_read, paste,
   open_app, open_tab, close_tab, browser_back, browser_forward, browser_search,
   screen_info, reset_computer. إزالة الكود الميت قلّلت التعقيد والـ WebSocket
   overhead لأدوات لا أحد يستدعيها.

── التوافق مع الواجهة الأمامية ──
   البروتوكول (أسماء actions، أسماء حقول رسائل screenshot/terminal_result/
   computer_reset/error/connected/screen_info) مطابق 100% لما تتوقعه
   index_plus_prompt_enhanced_v2-3.html — لا حاجة لأي تعديل في الفرونت إند.

── v20.1 — إضافات: تعدد المتصفحات + بحث رسمي بدون كابتشا ──
1) تعدد المتصفحات (Multi-Browser Fallback):
   عند إقلاع السيرفر، يُفحص كل مرشحي المتصفح المعروفين (Chromium بأسمائه
   المختلفة حسب التوزيعة، Chrome، Edge، Brave...) بتشغيل حقيقي فعلي (مو مجرد
   "هل موجود بالمسار") واختيار أول واحد يستجيب فعلاً على CDP. لو المتصفح
   المعتمد فشل لاحقاً أثناء تشغيل جلسة مستخدم حقيقية (مثلاً تحديث كسر الثنائي)،
   السيرفر يجرب بقية المرشحين تلقائياً ويعتمد أول بديل ناجح لبقية الجلسات —
   بدون أي تدخل يدوي أو إعادة نشر.

2) Brave Search API — قناة بحث رسمية بدون كابتشا إطلاقاً:
   المشكلة الأصلية: فتح duckduckgo.com/google.com كموقع فعلي عبر المتصفح
   لأجل "البحث عن معلومة" يواجه كابتشا حتمياً، لأن هذي المحركات مصممة لمنع
   الطلبات الآلية تحديداً. الحل: بدل زيارة الموقع، الذكاء الاصطناعي يرسل أمر
   terminal بصيغة   search: نص الاستعلام   (بدل curl معقد) فيُستدعى Brave
   Search API مباشرة (بدون فتح أي متصفح، بدون كابتشا، لأنه مرخّص لهذا الغرض
   تحديداً من مزوّده). فتح متصفح Computer Mode الفعلي يبقى محفوظاً فقط لما
   يحتاج المستخدم "تصفح موقع بعينه بصرياً" أو التفاعل مع صفحة معينة — بالضبط
   نفس فلسفة عمل وكلاء مثل Manus: بحث نظيف عبر API لسطر الأوامر أولاً، ومتصفح
   حقيقي فقط للتصفح الفعلي.

   التفعيل: عيّن متغيّر بيئة BRAVE_API_KEY (من
   https://api-dashboard.search.brave.com — فيه باقة مجانية 2000 استعلام/شهر)
   على منصة النشر (Render/HF Spaces → Settings → Environment Variables).
   بدون المفتاح، يستمر السيرفر بالعمل عبر مصادر API المجانية البديلة
   (DuckDuckGo Instant Answer, Wikipedia, HackerNews, Reddit) تلقائياً.

── v20.2 — Camoufox: محرك متصفح ثانٍ مضاد للكشف ──
   المشكلة: حتى مع Brave Search API، أحياناً المستخدم يطلب فعلياً "تصفّح
   موقع معيّن بصرياً" (مو بحث)، وبعض المواقع (خصوصاً المحمية بـ Cloudflare/
   DataDome) تكتشف Chromium العادي فوراً حتى لو headless، لأنه يسرّب علامات
   أتمتة واضحة (navigator.webdriver=true، بصمة toString مكشوفة، إلخ).

   الحل: محرك بديل اختياري (Camoufox — Firefox معدَّل من مستوى C++، مفتوح
   المصدر ومجاني بالكامل) يُخفي هذي العلامات وينتج بصمة متصفح واقعية إحصائياً.
   يُفعَّل عبر أمر جديد يرسله الفرونت إند:
       {"action": "switch_engine", "data": {"engine": "camoufox"}}
   وللرجوع للمحرك الافتراضي السريع:
       {"action": "switch_engine", "data": {"engine": "chromium"}}
   بقية كل الأوامر (screenshot, mouse_click, keyboard_type...) تبقى بنفس
   الصيغة والبروتوكول تماماً — فقط تُنفَّذ عبر المحرك النشط للمستخدم، بدون
   أي تعديل مطلوب على الفرونت إند لتوجيه الأوامر بنفسها.

   تنبيه واقعي مهم: Camoufox يقلل ظهور الكابتشا بشكل ملحوظ، لكنه **ليس حلاً
   سحرياً 100%** — مواثيق Camoufox نفسها تذكر أن مواقع متقدمة جداً قد تكتشفه
   أحياناً عبر تحليل بصمات متقدم. كما أن سمعة IP السيرفر (Render/HF) تبقى
   عاملاً منفصلاً؛ Camoufox يعالج بصمة *المتصفح*، لا بصمة *الشبكة/IP*.

   المتطلبات: pip install camoufox[geoip] playwright، ثم
   playwright install firefox (أو camoufox fetch لتنزيل الثنائي المعدَّل
   تلقائياً عند أول استخدام، ~150-300MB لمرة واحدة). بدون هذي الحزم، أوامر
   switch_engine إلى camoufox تفشل بأمان وتُبقي الجلسة على Chromium.
"""

import asyncio
import base64
import hashlib
import json
import os
import re
import subprocess
import threading
import time
import urllib.parse
from typing import Optional

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, HTMLResponse
import uvicorn

try:
    import websockets as ws_lib
    _WS_LIB_AVAILABLE = True
except ImportError:
    _WS_LIB_AVAILABLE = False
    print("⚠️ websockets غير مثبّتة — pip install websockets (مطلوبة لهذا الإصدار)")


# ════════════════════════════════════════════════════════════════
# ── الإعدادات العامة ───────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

# أقصى عدد جلسات متزامنة. رقم واقعي لسيرفر موارد محدودة (HF Free CPU
# Basic: 2 vCPU/16GB مشتركة) — كل جلسة = عملية Chromium واحدة فقط الآن
# (بدل Xvfb+متصفح سابقاً)، فالحمل الفعلي لكل مستخدم أخف من النصف تقريباً
# مقارنة بالبنية القديمة، لكن نبقي رقماً متحفظاً بدل الافتراض المتفائل.
MAX_SESSIONS = 40

CDP_PORT_BASE = 9300
SCREEN_W, SCREEN_H = 1280, 800

_sessions_lock = threading.Lock()
_cdp_port_pool_lock = threading.Lock()
_used_ports: set = set()


def _detect_chromium() -> str:
    """
    يكتشف أي متصفح متوفر على هذا السيرفر — يجرب كل الأسماء الشائعة (Chromium,
    Chrome, Chromium الرسمي بأشكاله المختلفة حسب توزيعة Linux)، مو اسم واحد بس.
    """
    return _pick_working_browser()


# كل الأسماء الشائعة لثنائيات المتصفح على توزيعات Linux المختلفة (Debian/Ubuntu،
# Alpine، RPM-based) — الذكاء الاصطناعي (عبر هذه الدالة) يجرب كل واحد بالترتيب
# ويستخدم أول واحد يعمل فعلياً، بدل الاعتماد على اسم واحد قد لا يوجد على صورة
# الحاوية المستخدمة.
_BROWSER_CANDIDATES = [
    "chromium-browser",      # Debian/Ubuntu
    "chromium",               # Alpine/Arch/Fedora
    "google-chrome-stable",   # Google Chrome الرسمي
    "google-chrome",
    "chromium-freeworld",     # بعض توزيعات RPM
    "ungoogled-chromium",     # نسخة خصوصية من Chromium
    "microsoft-edge-stable",  # Edge (مبني على Chromium، يدعم نفس CDP)
    "brave-browser",          # Brave (مبني على Chromium، يدعم نفس CDP)
]


def _binary_exists(name: str) -> bool:
    r = subprocess.run(["which", name], capture_output=True, text=True)
    return r.returncode == 0 and bool(r.stdout.strip())


def _test_browser_launch(binary: str, timeout: float = 6.0) -> bool:
    """
    اختبار حقيقي: يشغّل المتصفح فعلياً بوضع headless على منفذ CDP مؤقت
    ويتأكد أنه يستجيب — مو مجرد فحص "موجود بالمسار"، لأن أحياناً الثنائي
    موجود لكن مكسور (مكتبات ناقصة، إلخ).
    """
    test_port = CDP_PORT_BASE + 9999  # منفذ مخصص للاختبار فقط، لا يتعارض مع الجلسات
    try:
        subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={test_port}"],
                       capture_output=True, timeout=3)
    except Exception:
        pass
    try:
        proc = subprocess.Popen([
            binary, "--headless=new", f"--remote-debugging-port={test_port}",
            "--remote-debugging-address=127.0.0.1", "--disable-gpu", "--no-sandbox",
            "--disable-dev-shm-usage", "--no-first-run", "about:blank",
        ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except Exception:
        return False

    ok = False
    deadline = time.time() + timeout
    try:
        import urllib.request
        while time.time() < deadline:
            try:
                with urllib.request.urlopen(f"http://127.0.0.1:{test_port}/json/version", timeout=1.0) as r:
                    if r.status == 200:
                        ok = True
                        break
            except Exception:
                time.sleep(0.3)
    finally:
        try:
            proc.kill()
            proc.wait(timeout=3)
        except Exception:
            pass
        try:
            subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={test_port}"],
                           capture_output=True, timeout=3)
        except Exception:
            pass
    return ok


def _pick_working_browser() -> str:
    """
    يجرب كل مرشحي المتصفح بالترتيب: أولاً يتأكد أن الثنائي موجود على القرص
    (سريع)، وإذا وجد، يختبره فعلياً بتشغيله والتأكد أنه يستجيب على CDP
    (اختبار حقيقي، لا افتراض). أول متصفح "حي" فعلاً يُعتمد لبقية عمر السيرفر.
    """
    found_but_untested = []
    for name in _BROWSER_CANDIDATES:
        if _binary_exists(name):
            found_but_untested.append(name)

    if not found_but_untested:
        print("❌ لا يوجد أي متصفح مثبّت على هذا السيرفر من بين المرشحين المعروفين:")
        print(f"   {', '.join(_BROWSER_CANDIDATES)}")
        print("   ثبّت واحداً عبر: apt-get install chromium")
        return "chromium-browser"  # افتراضي، حتى لو غير موجود، ليكون اسم واضح بالسجلات

    print(f"🔍 متصفحات موجودة على القرص: {', '.join(found_but_untested)} — يجري اختبار كل واحد فعلياً...")
    for name in found_but_untested:
        print(f"   🧪 اختبار {name} ...")
        if _test_browser_launch(name):
            print(f"   ✅ {name} يعمل فعلياً — سيُعتمد لكل الجلسات")
            return name
        else:
            print(f"   ❌ {name} موجود لكن فشل بالتشغيل الفعلي (مكتبات ناقصة أو ثنائي معطوب)")

    # كل شيء وُجد لكن فشل الاختبار الحي — استخدم أول موجود كخيار أخير مع تحذير واضح
    print(f"⚠️ كل المتصفحات الموجودة فشلت باختبار التشغيل الحي. سيُستخدم '{found_but_untested[0]}' "
         f"كخيار افتراضي أخير، لكن يُنصح بفحص تثبيت المتصفح على هذا السيرفر.")
    return found_but_untested[0]


BROWSER_BIN = _detect_chromium()


def _alloc_cdp_port() -> int:
    """يخصص منفذ CDP حر لمستخدم جديد."""
    with _cdp_port_pool_lock:
        p = CDP_PORT_BASE
        while p in _used_ports:
            p += 1
        _used_ports.add(p)
        return p


def _release_cdp_port(port: int):
    with _cdp_port_pool_lock:
        _used_ports.discard(port)


# ════════════════════════════════════════════════════════════════
# ── Camoufox — محرك متصفح بديل مضاد للكشف (اختياري، بجانب Chromium) ──
# ════════════════════════════════════════════════════════════════
# لماذا محرك ثاني منفصل؟ Camoufox هو Firefox معدَّل من مستوى C++ (مو حقن
# JavaScript قابل للكشف) يُخفي علامات الأتمتة (navigator.webdriver وغيرها)
# ويولّد بصمة متصفح متّسقة وواقعية إحصائياً (نظام تشغيل، دقة شاشة، خطوط،
# WebGL...) عبر مكتبة BrowserForge. هذا يقلل ظهور كابتشا المواقع المحمية
# بشدة (Cloudflare, DataDome) بشكل ملحوظ — لكنه **ليس ضماناً مطلقاً** (حتى
# موثّق Camoufox نفسه يذكر أن مواقع متقدمة قد تكتشفه أحياناً).
#
# الفرق الجوهري عن Chromium/CDP: واجهة Camoufox الرسمية هي Python+Playwright
# (مو CDP خام)، فتحكمنا فيه يمر عبر Playwright API (page.mouse, page.keyboard,
# page.screenshot) بدل رسائل CDP JSON المباشرة المستخدمة مع Chromium. لذلك
# بنيناه كـ"محرك بديل" مستقل بجانب CDPSession، بنفس بروتوكول الأكشنات الخارجي
# (screenshot/mouse_click/keyboard_type...) حتى لا يتأثر الفرونت إند إطلاقاً.
#
# التفعيل: تلقائي بالكامل — لا حاجة لأي إعداد. Camoufox حزمة بايثون عادية
# (pip install camoufox[geoip] + playwright)، أول استخدام يُنزّل ثنائي
# Firefox المعدَّل تلقائياً (~150-300MB مرة واحدة فقط، يُخزَّن ويُعاد استخدامه).

try:
    from camoufox.async_api import AsyncCamoufox
    _CAMOUFOX_LIB_AVAILABLE = True
except ImportError:
    _CAMOUFOX_LIB_AVAILABLE = False
    print("ℹ️ مكتبة camoufox غير مثبّتة — pip install camoufox playwright لتفعيل المحرك المضاد للكشف")

CAMOUFOX_ENABLED = os.environ.get("CAMOUFOX_ENABLED", "1").strip() not in ("0", "false", "no")


class CamoufoxSession:
    """
    جلسة Camoufox واحدة لكل مستخدم — بديل لـ CDPSession لكن مبني فوق
    Playwright بدل CDP خام. يوفر نفس نوع العمليات (screenshot, mouse,
    keyboard, navigate) بواجهة موحّدة (dict نتائج) لتُستهلك من نفس معالج
    الأكشنات (handle_action) دون تغيير بروتوكول الفرونت إند.
    """

    def __init__(self, user_id: str):
        self.user_id = user_id
        self._cm = None       # AsyncCamoufox context manager instance
        self._browser = None  # Playwright Browser
        self._context = None  # Playwright BrowserContext
        self.page = None      # Playwright Page الحالية
        self._mouse_x = SCREEN_W // 2
        self._mouse_y = SCREEN_H // 2

    @property
    def alive(self) -> bool:
        return self.page is not None and not self.page.is_closed()

    async def start(self) -> bool:
        if not _CAMOUFOX_LIB_AVAILABLE:
            print("[camoufox] ❌ المكتبة غير مثبَّتة على هذا السيرفر")
            return False
        try:
            profile = _profile_dir("camoufox_" + self.user_id)
            self._cm = AsyncCamoufox(
                headless=True,                    # Camoufox يدعم headless حقيقي بأمان (يُخفي علامات الكشف حتى في هذا الوضع)
                humanize=True,                     # حركة فأرة شبيهة بالبشر (منحنيات Bezier بدل خطوط مستقيمة)
                geoip=False,                       # لا نحتاج ضبط جغرافي تلقائي عبر بروكسي هنا
                persistent_context=True,
                user_data_dir=profile,
                window=(SCREEN_W, SCREEN_H),
            )
            self._context = await self._cm.__aenter__()
            # في persistent_context=True، الكائن المُعاد هو BrowserContext مباشرة
            pages = self._context.pages
            self.page = pages[0] if pages else await self._context.new_page()
            await self.page.set_viewport_size({"width": SCREEN_W, "height": SCREEN_H})
            print(f"[camoufox] ✅ جلسة جديدة لـ '{self.user_id}' جاهزة")
            return True
        except Exception as e:
            print(f"[camoufox] ❌ فشل بدء الجلسة: {e}")
            await self.close()
            return False

    async def close(self):
        try:
            if self._cm:
                await self._cm.__aexit__(None, None, None)
        except Exception:
            pass
        self._cm = None
        self._context = None
        self.page = None

    async def screenshot_b64(self) -> "str | None":
        if not self.alive:
            return None
        try:
            raw = await self.page.screenshot(type="jpeg", quality=75)
            return base64.b64encode(raw).decode()
        except Exception as e:
            print(f"[camoufox] ❌ فشل التقاط الشاشة: {e}")
            return None

    async def mouse_move(self, x: int, y: int):
        if self.alive:
            await self.page.mouse.move(x, y)
        self._mouse_x, self._mouse_y = x, y

    async def mouse_click(self, x: int, y: int, button: str = "left", double: bool = False):
        if not self.alive:
            return
        await self.page.mouse.move(x, y)
        if double:
            await self.page.mouse.dblclick(x, y, button=button)
        else:
            await self.page.mouse.click(x, y, button=button)
        self._mouse_x, self._mouse_y = x, y

    async def mouse_drag(self, x1: int, y1: int, x2: int, y2: int):
        if not self.alive:
            return
        await self.page.mouse.move(x1, y1)
        await self.page.mouse.down()
        await self.page.mouse.move(x2, y2, steps=10)
        await self.page.mouse.up()
        self._mouse_x, self._mouse_y = x2, y2

    async def scroll(self, x: int, y: int, delta_y: int):
        if not self.alive:
            return
        await self.page.mouse.move(x, y)
        await self.page.mouse.wheel(0, delta_y)

    async def type_text(self, text: str):
        if self.alive:
            await self.page.keyboard.type(text, delay=25)  # تأخير خفيف بين الحروف — أكثر واقعية بشرياً

    async def key_press(self, key_name: str):
        if self.alive:
            await self.page.keyboard.press(key_name)

    async def hotkey(self, keys: list):
        if not self.alive:
            return
        combo = "+".join(keys)
        await self.page.keyboard.press(combo)

    async def navigate(self, url: str) -> bool:
        if not self.alive:
            return False
        try:
            await self.page.goto(url, timeout=20000, wait_until="domcontentloaded")
            return True
        except Exception as e:
            print(f"[camoufox] ⚠️ فشل التنقّل إلى {url}: {e}")
            return False

    async def go_back(self):
        if self.alive:
            try:
                await self.page.go_back(timeout=8000)
            except Exception:
                pass

    async def go_forward(self):
        if self.alive:
            try:
                await self.page.go_forward(timeout=8000)
            except Exception:
                pass


# جلسات Camoufox منفصلة عن جلسات Chromium/CDP — مفتاحها user_id أيضاً، لكن
# قاموس مستقل لأن محرك المستخدم قد يتبدّل بين الاثنين أثناء الجلسة نفسها.
_camoufox_sessions: dict = {}
_camoufox_lock = threading.Lock()


async def get_camoufox_session(user_id: str) -> "CamoufoxSession | None":
    """يُرجع جلسة Camoufox موجودة أو ينشئ واحدة جديدة عند الحاجة."""
    if not CAMOUFOX_ENABLED or not _CAMOUFOX_LIB_AVAILABLE:
        return None
    with _camoufox_lock:
        sess = _camoufox_sessions.get(user_id)
    if sess and sess.alive:
        return sess
    sess = CamoufoxSession(user_id)
    ok = await sess.start()
    if not ok:
        return None
    with _camoufox_lock:
        _camoufox_sessions[user_id] = sess
    return sess


async def destroy_camoufox_session(user_id: str):
    with _camoufox_lock:
        sess = _camoufox_sessions.pop(user_id, None)
    if sess:
        await sess.close()


# ════════════════════════════════════════════════════════════════
# ── CDP Session — الاتصال المباشر بـ Chromium عبر DevTools Protocol ──
# ════════════════════════════════════════════════════════════════

class CDPSession:
    """
    اتصال WebSocket واحد بجلسة Chromium headless واحدة. كل الأوامر
    (screenshot, mouse, keyboard, navigate) تمر من هنا. اتصال واحد يُعاد
    استخدامه طوال حياة الجلسة بدل فتح اتصال جديد لكل أمر.
    """

    def __init__(self, port: int):
        self.port = port
        self._ws = None
        self._msg_id = 0
        self._pending: dict = {}
        self._recv_task: Optional[asyncio.Task] = None
        self._send_lock = asyncio.Lock()
        self._target_id = None

    async def connect(self, timeout: float = 6.0) -> bool:
        if not _WS_LIB_AVAILABLE:
            return False
        # جرّب عدة مرات لأن Chromium قد يحتاج لحظة بعد الإقلاع قبل فتح منفذ CDP
        deadline = time.time() + timeout
        tabs = None
        while time.time() < deadline:
            try:
                import urllib.request
                with urllib.request.urlopen(
                    f"http://127.0.0.1:{self.port}/json/list", timeout=1.5
                ) as r:
                    tabs = json.loads(r.read())
                break
            except Exception:
                await asyncio.sleep(0.25)
        if tabs is None:
            print(f"[CDP:{self.port}] ❌ Chromium لا يستجيب على منفذ CDP")
            return False

        ws_url = None
        for t in tabs:
            if t.get("type") == "page":
                ws_url = t.get("webSocketDebuggerUrl")
                self._target_id = t.get("id")
                break
        if not ws_url:
            print(f"[CDP:{self.port}] ❌ لا يوجد تبويب صفحة متاح")
            return False

        try:
            self._ws = await asyncio.wait_for(
                ws_lib.connect(ws_url, ping_interval=15, ping_timeout=10, max_size=64 * 1024 * 1024),
                timeout=timeout,
            )
        except Exception as e:
            print(f"[CDP:{self.port}] ❌ فشل اتصال WebSocket: {e}")
            return False

        self._recv_task = asyncio.create_task(self._recv_loop())

        # فعّل النطاقات المطلوبة
        await self.send("Page.enable", timeout=5.0)
        await self.send("Runtime.enable", timeout=5.0)
        await self.send("DOM.enable", timeout=5.0)
        # اضبط حجم الشاشة الافتراضي عبر Emulation (headless بدون نافذة فعلية،
        # هذا يحدد أبعاد الرندر الداخلي)
        await self.send("Emulation.setDeviceMetricsOverride", {
            "width": SCREEN_W, "height": SCREEN_H,
            "deviceScaleFactor": 1, "mobile": False,
        }, timeout=5.0)
        print(f"[CDP:{self.port}] ✅ متصل بجلسة Chromium")
        return True

    async def _recv_loop(self):
        try:
            async for raw in self._ws:
                try:
                    msg = json.loads(raw)
                except Exception:
                    continue
                mid = msg.get("id")
                if mid is not None and mid in self._pending:
                    fut = self._pending.pop(mid)
                    if not fut.done():
                        fut.set_result(msg)
        except Exception:
            pass
        finally:
            for fut in self._pending.values():
                if not fut.done():
                    fut.set_exception(ConnectionError("CDP disconnected"))
            self._pending.clear()
            self._ws = None

    async def send(self, method: str, params: dict = None, timeout: float = 10.0):
        if not self._ws:
            return None
        async with self._send_lock:
            self._msg_id += 1
            mid = self._msg_id
        fut = asyncio.get_event_loop().create_future()
        self._pending[mid] = fut
        try:
            await self._ws.send(json.dumps({"id": mid, "method": method, "params": params or {}}))
            return await asyncio.wait_for(fut, timeout=timeout)
        except Exception:
            self._pending.pop(mid, None)
            return None

    @property
    def alive(self) -> bool:
        return self._ws is not None

    async def close(self):
        if self._recv_task:
            self._recv_task.cancel()
        if self._ws:
            try:
                await self._ws.close()
            except Exception:
                pass
        self._ws = None


# ════════════════════════════════════════════════════════════════
# ── تشغيل/إيقاف عملية Chromium headless ────────────────────────
# ════════════════════════════════════════════════════════════════

def _profile_dir(user_id: str) -> str:
    safe = re.sub(r"[^a-zA-Z0-9_.-]", "_", user_id) or "anon"
    d = os.path.expanduser(f"~/.zai_chromium_profiles/{safe}")
    os.makedirs(d, exist_ok=True)
    return d


def _launch_browser_process(binary: str, port: int, profile: str) -> "subprocess.Popen | None":
    cmd = [
        binary,
        "--headless=new",                 # الوضع الرسمي الحديث — لا حاجة لأي display
        f"--remote-debugging-port={port}",
        "--remote-debugging-address=127.0.0.1",
        "--disable-gpu",
        "--no-sandbox",
        "--disable-dev-shm-usage",
        "--disable-extensions",
        "--disable-notifications",
        "--disable-popup-blocking",
        "--disable-translate",
        "--disable-sync",
        "--disable-background-timer-throttling",
        "--disable-backgrounding-occluded-windows",
        "--disable-renderer-backgrounding",
        "--disable-features=Translate,BackForwardCache",
        "--no-first-run",
        "--no-default-browser-check",
        "--mute-audio",
        "--hide-scrollbars",
        f"--window-size={SCREEN_W},{SCREEN_H}",
        f"--user-data-dir={profile}",
        "--js-flags=--max-old-space-size=256",  # سقف ذاكرة JS — يمنع تبويب ثقيل من استهلاك كل الرام
        "about:blank",
    ]
    try:
        proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return proc
    except FileNotFoundError:
        return None
    except Exception as e:
        print(f"[chromium] ❌ فشل تشغيل {binary}: {e}")
        return None


def _start_chromium(port: int, user_id: str) -> "subprocess.Popen | None":
    """
    يُشغّل متصفحاً بوضع headless=new على منفذ CDP مخصص. لا Xvfb، لا DISPLAY،
    لا نافذة — الرندر بالكامل داخلي (in-process) لمحرك المتصفح نفسه.

    ذكاء إضافي: لو المتصفح الحالي (BROWSER_BIN) فشل بالإقلاع فعلياً وقت التشغيل
    الحقيقي (مو وقت الفحص الأولي)، يجرّب بقية المتصفحات المرشحة تلقائياً بدل
    الفشل الكامل، ويحدّث BROWSER_BIN للمتصفح اللي نجح ليُستخدم مباشرة بالجلسات
    القادمة (بدون إعادة تجربة كل المتصفحات الفاشلة من جديد كل مرة).
    """
    global BROWSER_BIN
    profile = _profile_dir(user_id)
    try:
        subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={port}"],
                       capture_output=True, timeout=3)
        time.sleep(0.2)
    except Exception:
        pass

    proc = _launch_browser_process(BROWSER_BIN, port, profile)
    if proc is not None:
        print(f"[chromium] 🚀 بدأ '{BROWSER_BIN}' (PID {proc.pid}) على منفذ CDP {port} — headless=new")
        return proc

    print(f"[chromium] ⚠️ فشل '{BROWSER_BIN}' فعلياً — أجرب بقية المتصفحات المعروفة تلقائياً...")
    for alt in _BROWSER_CANDIDATES:
        if alt == BROWSER_BIN or not _binary_exists(alt):
            continue
        proc = _launch_browser_process(alt, port, profile)
        if proc is not None:
            print(f"[chromium] ✅ نجح البديل '{alt}' — سيُعتمد لبقية الجلسات من الآن")
            BROWSER_BIN = alt
            return proc

    print(f"[chromium] ❌ فشلت كل المتصفحات المتاحة على هذا السيرفر")
    return None


def _kill_chromium(proc: "subprocess.Popen | None", port: int):
    if proc:
        try:
            proc.kill()
            proc.wait(timeout=3)
        except Exception:
            pass
    try:
        subprocess.run(["pkill", "-9", "-f", f"[r]emote-debugging-port={port}"],
                       capture_output=True, timeout=3)
    except Exception:
        pass


# ════════════════════════════════════════════════════════════════
# ── إدارة الجلسات (مستخدم → CDPSession + Chromium process) ─────
# ════════════════════════════════════════════════════════════════

# user_id → {
#   "port": int, "proc": Popen, "cdp": CDPSession|None,
#   "active_ws": WebSocket|None, "created": float,
#   "consecutive_failures": int, "last_frame_hash": str,
#   "last_mouse_x": int, "last_mouse_y": int,
# }
_sessions: dict = {}


async def get_or_create_session(user_id: str, ws: WebSocket) -> dict:
    with _sessions_lock:
        sess = _sessions.get(user_id)
        if sess:
            sess["active_ws"] = ws
            print(f"[session] 🔁 إعادة اتصال '{user_id}' على منفذ {sess['port']}")
            return sess

        if len(_sessions) >= MAX_SESSIONS:
            # أعد استخدام أقدم جلسة غير متصلة بدل رفض المستخدم الجديد
            idle = [(uid, s) for uid, s in _sessions.items() if not s.get("active_ws")]
            if idle:
                oldest_uid = min(idle, key=lambda t: t[1]["created"])[0]
                old = _sessions.pop(oldest_uid)
                _kill_chromium(old.get("proc"), old["port"])
                _release_cdp_port(old["port"])
                print(f"[session] ♻️ أُزيلت جلسة خاملة قديمة '{oldest_uid}' لإفساح المجال")

        port = _alloc_cdp_port()
        sess = {
            "user_id": user_id,
            "port": port,
            "proc": None,
            "cdp": None,
            "active_ws": ws,
            "created": time.time(),
            "consecutive_failures": 0,
            "last_frame_hash": "",
            "last_mouse_x": SCREEN_W // 2,
            "last_mouse_y": SCREEN_H // 2,
            "shot_lock": asyncio.Lock(),
            "engine": "chromium",   # "chromium" (افتراضي، عبر CDP) أو "camoufox" (مضاد للكشف، عبر Playwright)
        }
        _sessions[user_id] = sess
        print(f"[session] ✅ مستخدم جديد '{user_id}' → منفذ CDP {port}")
        return sess


async def ensure_browser_running(sess: dict) -> bool:
    """يتأكد أن Chromium + اتصال CDP يعملان لهذه الجلسة. يُشغّلهما إذا لزم."""
    if sess.get("cdp") and sess["cdp"].alive:
        return True

    if not sess.get("proc") or sess["proc"].poll() is not None:
        proc = await asyncio.to_thread(_start_chromium, sess["port"], sess["user_id"])
        sess["proc"] = proc
        if proc is None:
            return False

    cdp = CDPSession(sess["port"])
    ok = await cdp.connect(timeout=8.0)
    if ok:
        sess["cdp"] = cdp
        return True
    return False


async def reset_session(user_id: str) -> bool:
    """يُعيد تشغيل Chromium بالكامل لهذا المستخدم — الإصلاح الشامل الوحيد المطلوب الآن،
    بما أنه لا يوجد Xvfb أو ملفات جلسة X منفصلة يجب تنظيفها."""
    with _sessions_lock:
        sess = _sessions.get(user_id)
        if not sess:
            return False
        proc, port = sess.get("proc"), sess["port"]
        old_cdp = sess.get("cdp")

    if old_cdp:
        await old_cdp.close()
    await asyncio.to_thread(_kill_chromium, proc, port)

    new_proc = await asyncio.to_thread(_start_chromium, port, user_id)
    new_cdp = CDPSession(port)
    ok = await new_cdp.connect(timeout=8.0) if new_proc else False

    with _sessions_lock:
        if user_id in _sessions:
            _sessions[user_id]["proc"] = new_proc
            _sessions[user_id]["cdp"] = new_cdp if ok else None
            _sessions[user_id]["consecutive_failures"] = 0
            _sessions[user_id]["last_frame_hash"] = ""
    print(f"[reset] {'✅' if ok else '❌'} إعادة ضبط '{user_id}' على منفذ {port}")
    return ok


async def destroy_session_ws(user_id: str, ws: WebSocket):
    """عند انقطاع WebSocket — نُبقي Chromium شغالاً (الجلسة محفوظة) وفقط نزيل الاتصال."""
    with _sessions_lock:
        sess = _sessions.get(user_id)
        if sess and sess.get("active_ws") is ws:
            sess["active_ws"] = None
    print(f"[session] 📴 '{user_id}' قطع الاتصال (الجلسة محفوظة)")


async def destroy_session_full(user_id: str):
    with _sessions_lock:
        sess = _sessions.pop(user_id, None)
    if sess:
        if sess.get("cdp"):
            await sess["cdp"].close()
        await asyncio.to_thread(_kill_chromium, sess.get("proc"), sess["port"])
        _release_cdp_port(sess["port"])
    await destroy_camoufox_session(user_id)


# ════════════════════════════════════════════════════════════════
# ── لقطة الشاشة — عبر CDP مباشرة (بدون subprocess خارجي) ────────
# ════════════════════════════════════════════════════════════════

async def cdp_screenshot(sess: dict, quality: int = 75) -> "tuple[str, int, int] | None":
    """
    يلتقط لقطة شاشة عبر Page.captureScreenshot — أمر CDP واحد يرجع JPEG
    base64 جاهز مباشرة من محرك الرندر الداخلي لـ Chromium. لا يوجد X server
    يمكن أن "يتجمد"، لا حاجة لفحص "هل الصورة سوداء"، ولا عملية subprocess
    خارجية (import/xwd) قد تفشل بسبب أدوات مفقودة على السيرفر.
    يُعيد (base64_jpeg, width, height) أو None عند الفشل.
    """
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return None
    result = await cdp.send("Page.captureScreenshot", {
        "format": "jpeg", "quality": quality, "fromSurface": True, "captureBeyondViewport": False,
    }, timeout=8.0)
    if not result:
        return None
    data = result.get("result", {}).get("data")
    if not data:
        return None
    return data, SCREEN_W, SCREEN_H


def _draw_mouse_marker_on_jpeg(b64_jpeg: str, mx: int, my: int) -> str:
    """يرسم علامة صغيرة لموقع الفأرة فوق الصورة (مطلوبة بصرياً للذكاء الاصطناعي
    ليرى أين المؤشر). العملية خفيفة (رسم فقط، بدون resize) فتُبقي زمن كل لقطة منخفضاً."""
    try:
        from PIL import Image, ImageDraw
        import io as _io
        raw = base64.b64decode(b64_jpeg)
        img = Image.open(_io.BytesIO(raw)).convert("RGB")
        draw = ImageDraw.Draw(img)
        r = 9
        draw.ellipse([(mx - r, my - r), (mx + r, my + r)], outline=(255, 60, 60), width=2)
        draw.line([(mx - 14, my), (mx + 14, my)], fill=(255, 60, 60), width=1)
        draw.line([(mx, my - 14), (mx, my + 14)], fill=(255, 60, 60), width=1)
        buf = _io.BytesIO()
        img.save(buf, format="JPEG", quality=80, optimize=True)
        return base64.b64encode(buf.getvalue()).decode()
    except Exception:
        return b64_jpeg  # لو فشل الرسم لأي سبب، أعد الصورة الأصلية بدل الفشل الكامل


def _grid_variant(b64_jpeg: str, grid_step: int, color: tuple, label_every: int) -> str:
    """
    ينتج نسخة من الصورة مع شبكة إحداثيات مرسومة فوقها — تساعد الذكاء الاصطناعي
    على تحديد إحداثيات دقيقة للنقر. نفس فكرة v13 لكن مبنية فوق JPEG الجاهز من
    CDP مباشرة بدل إعادة الرسم من raw pixels.
    """
    try:
        from PIL import Image, ImageDraw
        import io as _io
        raw = base64.b64decode(b64_jpeg)
        img = Image.open(_io.BytesIO(raw)).convert("RGB")
        draw = ImageDraw.Draw(img, "RGBA")
        w, h = img.size
        major = (*color, 130)
        minor = (*color, 40)
        x = grid_step
        while x < w:
            is_major = x % label_every == 0
            draw.line([(x, 0), (x, h)], fill=major if is_major else minor, width=1)
            if is_major:
                draw.rectangle([(x + 1, 2), (x + 34, 15)], fill=(0, 0, 0, 175))
                draw.text((x + 2, 3), str(x), fill=color)
            x += grid_step
        y = grid_step
        while y < h:
            is_major = y % label_every == 0
            draw.line([(0, y), (w, y)], fill=major if is_major else minor, width=1)
            if is_major:
                draw.rectangle([(2, y + 1), (38, y + 14)], fill=(0, 0, 0, 175))
                draw.text((3, y + 2), str(y), fill=color)
            y += grid_step
        final = img.convert("RGB")
        buf = _io.BytesIO()
        final.save(buf, format="JPEG", quality=78, optimize=True)
        return base64.b64encode(buf.getvalue()).decode()
    except Exception:
        return b64_jpeg


async def capture_full(sess: dict, quality: int = 75) -> dict:
    """
    يلتقط لقطة واحدة وينتج منها: نسخة نظيفة + نسخة بعلامة الفأرة + شبكتان
    إحداثيات (عادية ودقيقة) — بنفس تنسيق حقول الرد الذي تتوقعه الواجهة الأمامية
    (data, data_grid, data_grid2, data_grid3, width, height, mouse_x, mouse_y).

    يعمل مع كلا المحركين حسب sess["engine"]:
      "chromium" (افتراضي) → عبر CDP مباشرة (cdp_screenshot)
      "camoufox" → عبر Playwright screenshot (أبطأ قليلاً لكن مضاد للكشف)
    """
    if sess.get("engine") == "camoufox":
        cam = await get_camoufox_session(sess["user_id"])
        if not cam:
            return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "",
                    "width": SCREEN_W, "height": SCREEN_H,
                    "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0),
                    "is_placeholder": True}
        raw_b64 = await cam.screenshot_b64()
        if not raw_b64:
            return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "",
                    "width": SCREEN_W, "height": SCREEN_H,
                    "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0),
                    "is_placeholder": True}
        w, h = SCREEN_W, SCREEN_H
        mx, my = sess.get("last_mouse_x", w // 2), sess.get("last_mouse_y", h // 2)

        def _build_cam():
            clean = _draw_mouse_marker_on_jpeg(raw_b64, mx, my)
            grid1 = _grid_variant(clean, 50, (0, 255, 180), 100)
            grid2 = _grid_variant(clean, 20, (255, 60, 60), 60)
            return clean, grid1, grid2

        clean, grid1, grid2 = await asyncio.to_thread(_build_cam)
        return {"data": clean, "data_grid": grid1, "data_grid2": grid2, "data_grid3": grid2,
                "width": w, "height": h, "mouse_x": mx, "mouse_y": my, "is_placeholder": False}

    shot = await cdp_screenshot(sess, quality)
    if not shot:
        return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "",
                "width": SCREEN_W, "height": SCREEN_H,
                "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0),
                "is_placeholder": True}

    raw_b64, w, h = shot
    mx, my = sess.get("last_mouse_x", w // 2), sess.get("last_mouse_y", h // 2)

    def _build():
        clean = _draw_mouse_marker_on_jpeg(raw_b64, mx, my)
        grid1 = _grid_variant(clean, 50, (0, 255, 180), 100)
        grid2 = _grid_variant(clean, 20, (255, 60, 60), 60)
        return clean, grid1, grid2

    clean, grid1, grid2 = await asyncio.to_thread(_build)
    return {
        "data": clean, "data_grid": grid1, "data_grid2": grid2, "data_grid3": grid2,
        "width": w, "height": h, "mouse_x": mx, "mouse_y": my,
        "is_placeholder": False,
    }


def _frame_hash(data: str) -> str:
    return hashlib.md5(data.encode()).hexdigest()


# ════════════════════════════════════════════════════════════════
# ── فأرة/كيبورد — عبر CDP Input domain (بدون xdotool/X server) ──
# ════════════════════════════════════════════════════════════════

async def cdp_mouse_move(sess: dict, x: int, y: int):
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    await cdp.send("Input.dispatchMouseEvent", {
        "type": "mouseMoved", "x": x, "y": y,
    }, timeout=4.0)
    sess["last_mouse_x"], sess["last_mouse_y"] = x, y


async def cdp_mouse_click(sess: dict, x: int, y: int, button: str = "left", double: bool = False):
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    await cdp_mouse_move(sess, x, y)
    click_count = 2 if double else 1
    await cdp.send("Input.dispatchMouseEvent", {
        "type": "mousePressed", "x": x, "y": y, "button": button, "clickCount": click_count,
    }, timeout=4.0)
    await asyncio.sleep(0.04)
    await cdp.send("Input.dispatchMouseEvent", {
        "type": "mouseReleased", "x": x, "y": y, "button": button, "clickCount": click_count,
    }, timeout=4.0)


async def cdp_mouse_drag(sess: dict, x1: int, y1: int, x2: int, y2: int, steps: int = 8):
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    await cdp_mouse_move(sess, x1, y1)
    await cdp.send("Input.dispatchMouseEvent", {
        "type": "mousePressed", "x": x1, "y": y1, "button": "left", "clickCount": 1,
    }, timeout=4.0)
    for i in range(1, steps + 1):
        ix = x1 + (x2 - x1) * i // steps
        iy = y1 + (y2 - y1) * i // steps
        await cdp.send("Input.dispatchMouseEvent", {
            "type": "mouseMoved", "x": ix, "y": iy, "button": "left",
        }, timeout=3.0)
        await asyncio.sleep(0.02)
    await cdp.send("Input.dispatchMouseEvent", {
        "type": "mouseReleased", "x": x2, "y": y2, "button": "left", "clickCount": 1,
    }, timeout=4.0)
    sess["last_mouse_x"], sess["last_mouse_y"] = x2, y2


async def cdp_scroll(sess: dict, x: int, y: int, delta_y: int):
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    await cdp.send("Input.dispatchMouseEvent", {
        "type": "mouseWheel", "x": x, "y": y, "deltaX": 0, "deltaY": delta_y,
    }, timeout=4.0)


# خريطة أسماء مفاتيح شائعة → أكواد CDP (windowsVirtualKeyCode + key)
_KEY_MAP = {
    "Return": (13, "Enter"), "Enter": (13, "Enter"),
    "Tab": (9, "Tab"), "Escape": (27, "Escape"), "BackSpace": (8, "Backspace"),
    "Delete": (46, "Delete"), "Left": (37, "ArrowLeft"), "Right": (39, "ArrowRight"),
    "Up": (38, "ArrowUp"), "Down": (40, "ArrowDown"), "Home": (36, "Home"), "End": (35, "End"),
    "space": (32, " "), "F5": (116, "F5"),
}


async def cdp_key_press(sess: dict, key_name: str):
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    code, key = _KEY_MAP.get(key_name, (0, key_name))
    params = {"type": "keyDown", "key": key}
    if code:
        params["windowsVirtualKeyCode"] = code
        params["nativeVirtualKeyCode"] = code
    await cdp.send("Input.dispatchKeyEvent", params, timeout=4.0)
    params["type"] = "keyUp"
    await cdp.send("Input.dispatchKeyEvent", params, timeout=4.0)


async def cdp_hotkey(sess: dict, keys: list):
    """اختصار متعدد المفاتيح مثل ['ctrl','l'] أو ['alt','Left']."""
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    modifiers = 0
    mod_map = {"alt": 1, "ctrl": 2, "control": 2, "meta": 4, "shift": 8}
    plain_keys = []
    for k in keys:
        kl = k.lower()
        if kl in mod_map:
            modifiers |= mod_map[kl]
        else:
            plain_keys.append(k)
    for k in plain_keys:
        code, key = _KEY_MAP.get(k, (ord(k[0].upper()) if len(k) == 1 else 0, k))
        await cdp.send("Input.dispatchKeyEvent", {
            "type": "rawKeyDown", "key": key, "modifiers": modifiers,
            "windowsVirtualKeyCode": code, "nativeVirtualKeyCode": code,
        }, timeout=4.0)
        await cdp.send("Input.dispatchKeyEvent", {
            "type": "keyUp", "key": key, "modifiers": modifiers,
            "windowsVirtualKeyCode": code, "nativeVirtualKeyCode": code,
        }, timeout=4.0)


async def cdp_type_text(sess: dict, text: str):
    """
    يكتب نصاً عبر Input.insertText — يدعم كل اللغات (عربي/إنجليزي/فرنسي)
    بشكل أصلي عبر CDP نفسه، بدون حاجة لـ xclip أو حيلة clipboard+paste
    التي كانت ضرورية سابقاً بسبب قيود xdotool مع النصوص غير اللاتينية.
    """
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    await cdp.send("Input.insertText", {"text": text}, timeout=6.0)


async def cdp_navigate(sess: dict, url: str) -> bool:
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return False
    result = await cdp.send("Page.navigate", {"url": url}, timeout=15.0)
    return bool(result and "error" not in result.get("result", {}))


async def cdp_clipboard_write(sess: dict, text: str):
    """ينسخ نصاً لحافظة afterward-paste داخلية عبر متغيّر JS بسيط (بديل خفيف
    لـ xclip، لأنه لا يوجد X server هنا أصلاً تُبنى عليه حافظة نظام حقيقية)."""
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return
    js = f"window.__zclip = {json.dumps(text)};"
    await cdp.send("Runtime.evaluate", {"expression": js}, timeout=4.0)


async def cdp_clipboard_read(sess: dict) -> str:
    cdp = sess.get("cdp")
    if not cdp or not cdp.alive:
        return ""
    result = await cdp.send("Runtime.evaluate", {
        "expression": "window.__zclip || ''", "returnByValue": True,
    }, timeout=4.0)
    if not result:
        return ""
    return result.get("result", {}).get("result", {}).get("value", "") or ""


async def cdp_get_screen_info(sess: dict) -> dict:
    return {"width": SCREEN_W, "height": SCREEN_H,
            "mouse_x": sess.get("last_mouse_x", 0), "mouse_y": sess.get("last_mouse_y", 0)}


# ════════════════════════════════════════════════════════════════
# ── SafeSearch — نفس منطق v13 محفوظاً كما هو (يعمل بشكل جيد) ────
# ════════════════════════════════════════════════════════════════

_BING_RE = re.compile(r"https?://(?:www\.)?bing\.com[^\s'\"]*")
_DDG_RE = re.compile(r"https?://(?:www\.)?duckduckgo\.com[^\s'\"]*")
_GOOG_RE = re.compile(r"https?://(?:www\.)?google\.[a-z.]+[^\s'\"]*")


def _safe_search(text: str) -> str:
    def _bing(m):
        u = m.group(0)
        return re.sub(r"adlt=\w+", "adlt=strict", u) if "adlt=" in u else u + ("&" if "?" in u else "?") + "adlt=strict"

    def _ddg(m):
        u = m.group(0)
        return re.sub(r"kp=\d", "kp=1", u) if "kp=" in u else u + ("&" if "?" in u else "?") + "kp=1"

    def _goog(m):
        u = m.group(0)
        return re.sub(r"safe=\w+", "safe=strict", u) if "safe=" in u else u + ("&" if "?" in u else "?") + "safe=strict"

    return _GOOG_RE.sub(_goog, _DDG_RE.sub(_ddg, _BING_RE.sub(_bing, text)))


# ════════════════════════════════════════════════════════════════
# ── Terminal — تنفيذ أوامر bash (منفصل تماماً عن Chromium) ──────
# ════════════════════════════════════════════════════════════════

_terminal_sem = asyncio.Semaphore(6)
_PKILL_RE = re.compile(r"\b(pkill|killall)\s+(-9\s+)?-f\s+(['\"]?)([a-zA-Z0-9_./-]+)\3")


def _sanitize_pkill(cmd: str) -> str:
    def _fix(m):
        tool, d9, _q, p = m.group(1), m.group(2) or "", m.group(3), m.group(4)
        sp = f"[{p[0]}]{p[1:]}" if len(p) > 1 else p
        return f"{tool} {d9}-f '{sp}'"
    return _PKILL_RE.sub(_fix, cmd)


async def run_cmd(cmd: str, timeout: int = 60) -> dict:
    cmd = _sanitize_pkill(_safe_search(cmd))
    async with _terminal_sem:
        env = {**os.environ, "PYTHONIOENCODING": "utf-8", "LANG": "en_US.UTF-8"}

        def _exec():
            try:
                r = subprocess.run(cmd, shell=True, capture_output=True, text=True,
                                   timeout=timeout, env=env, executable="/bin/bash")
                return {"stdout": r.stdout[-15000:], "stderr": r.stderr[-3000:], "returncode": r.returncode}
            except subprocess.TimeoutExpired:
                return {"stdout": "", "stderr": f"⏱️ Timeout {timeout}s", "returncode": -1}
            except Exception as e:
                return {"stdout": "", "stderr": str(e), "returncode": -1}

        return await asyncio.to_thread(_exec)


def _search_sources(query: str) -> list:
    q = urllib.parse.quote_plus(query)
    return [
        {"name": "DuckDuckGo Instant",
         "cmd": f"curl -s --max-time 15 'https://api.duckduckgo.com/?q={q}&format=json&no_html=1&skip_disambig=1' | python3 -c \"import sys,json;d=json.load(sys.stdin);a=d.get('AbstractText','');r=d.get('RelatedTopics',[]);print('ANS:',a or 'none');[print('-',x.get('Text','')[:200]) for x in r[:6] if isinstance(x,dict)]\""},
        {"name": "Wikipedia EN",
         "cmd": f"curl -s --max-time 12 'https://en.wikipedia.org/api/rest_v1/page/summary/{q}' | python3 -c \"import sys,json;d=json.load(sys.stdin);print(d.get('title','')+'\\n'+d.get('extract','')[:1200])\""},
        {"name": "DuckDuckGo HTML",
         "cmd": f"curl -sL --max-time 15 -H 'User-Agent: Mozilla/5.0' 'https://html.duckduckgo.com/html/?q={q}' | python3 -c \"import sys,re;h=sys.stdin.read();s=re.findall(r'class=.result__snippet[^>]*>(.*?)</a>',h,re.DOTALL);clean=lambda x:re.sub('<[^>]+>','',x).strip();[print(str(i+1)+'. '+clean(x)[:200]) for i,x in enumerate(s[:7])]\""},
        {"name": "HackerNews",
         "cmd": f"curl -s --max-time 12 'https://hn.algolia.com/api/v1/search?query={q}&hitsPerPage=6&tags=story' | python3 -c \"import sys,json;d=json.load(sys.stdin);[print(str(i+1)+'. '+h.get('title','')+'  Pts:'+str(h.get('points',0))) for i,h in enumerate(d.get('hits',[])[:5])]\""},
        {"name": "Reddit",
         "cmd": f"curl -sL --max-time 12 -H 'Accept: application/json' 'https://www.reddit.com/search.json?q={q}&sort=new&limit=6' | python3 -c \"import sys,json;d=json.load(sys.stdin);[print(str(i+1)+'. '+p['data'].get('title','')[:160]) for i,p in enumerate(d.get('data',{{}}).get('children',[])[:5])]\""},
    ]


def _result_empty(s: str) -> bool:
    if not s or len(s.strip()) < 15:
        return True
    return "ans: none" in s.lower()


# ════════════════════════════════════════════════════════════════
# ── Brave Search API — قناة البحث الرسمية المصرَّح بها ──────────
# ════════════════════════════════════════════════════════════════
# على عكس زيارة duckduckgo.com/google.com كموقع فعلي عبر المتصفح (اللي يواجه
# كابتشا لأن هذي المحركات تكره الطلبات الآلية)، Brave Search API مصمم أصلاً
# ليُستخدم برمجياً — بدون كابتشا إطلاقاً، لأنه مرخّص لهذا الغرض تحديداً.
# هذا بالضبط أسلوب Manus: بحث نظيف عبر API/سطر أوامر يرجع نتائج نصية مباشرة،
# فتح متصفح فعلي يُحفظ فقط لما يحتاج المستخدم "تصفح موقع بعينه بصرياً".
#
# التفعيل: عيّن متغيّر البيئة BRAVE_API_KEY بمفتاحك من
# https://api-dashboard.search.brave.com (فيه tier مجاني — 2000 استعلام/شهر).

BRAVE_API_KEY = os.environ.get("BRAVE_API_KEY", "").strip()
BRAVE_ENDPOINT = "https://api.search.brave.com/res/v1/web/search"


async def brave_search(query: str, count: int = 8) -> dict:
    """
    يستدعي Brave Search API مباشرة (بدون متصفح، بدون كابتشا). يُعيد نتائج
    منسّقة نصياً بنفس شكل مخرجات run_cmd العادية، عشان تندمج بسلاسة مع بقية
    منطق الترمنال والفرونت إند بدون أي تعديل إضافي مطلوب هناك.
    """
    if not BRAVE_API_KEY:
        return {"stdout": "", "stderr": "BRAVE_API_KEY غير مُعرَّف — راجع الإعدادات", "returncode": -1}

    def _fetch():
        import urllib.request
        import urllib.error
        url = f"{BRAVE_ENDPOINT}?q={urllib.parse.quote_plus(query)}&count={min(count, 20)}&safesearch=moderate"
        req = urllib.request.Request(url, headers={
            "Accept": "application/json",
            "Accept-Encoding": "gzip",
            "X-Subscription-Token": BRAVE_API_KEY,
        })
        try:
            with urllib.request.urlopen(req, timeout=15) as r:
                raw = r.read()
                if r.info().get("Content-Encoding") == "gzip":
                    import gzip
                    raw = gzip.decompress(raw)
                return json.loads(raw), None
        except urllib.error.HTTPError as e:
            body = e.read().decode(errors="ignore")[:300]
            return None, f"HTTP {e.code}: {body}"
        except Exception as e:
            return None, str(e)

    data, err = await asyncio.to_thread(_fetch)
    if err:
        return {"stdout": "", "stderr": f"Brave Search error: {err}", "returncode": -1}

    lines = []
    # الإجابة السريعة (infobox) إن وُجدت
    infobox = data.get("infobox", {}) or {}
    if infobox.get("long_desc") or infobox.get("description"):
        lines.append("ANS: " + (infobox.get("long_desc") or infobox.get("description", ""))[:400])

    web_results = (data.get("web", {}) or {}).get("results", []) or []
    for i, r in enumerate(web_results[:count], 1):
        title = r.get("title", "")
        desc = re.sub(r"<[^>]+>", "", r.get("description", ""))
        url = r.get("url", "")
        lines.append(f"{i}. {title}\n   {desc[:220]}\n   {url}")

    if not lines:
        return {"stdout": "", "stderr": "لا توجد نتائج من Brave Search لهذا الاستعلام", "returncode": 0}

    return {"stdout": "\n".join(lines)[:15000], "stderr": "", "returncode": 0}


# نمط يلتقط أمر بحث "زائف" مخصص يُرسله الذكاء الاصطناعي عبر أداة terminal
# بصيغة بسيطة: search: نص الاستعلام   (أو  search "نص الاستعلام")
_SEARCH_CMD_RE = re.compile(r"^\s*search\s*:?\s*[\"']?(.+?)[\"']?\s*$", re.IGNORECASE)


async def run_cmd_smart(cmd: str, timeout: int = 60) -> dict:
    # ── أولوية قصوى: أمر بحث صريح "search: ..." → Brave Search API مباشرة ──
    m_search = _SEARCH_CMD_RE.match(cmd)
    if m_search and "curl" not in cmd and "http" not in cmd:
        query = m_search.group(1).strip()
        if BRAVE_API_KEY:
            res = await brave_search(query)
            if res["returncode"] == 0 and not _result_empty(res["stdout"]):
                return res
            # فشل Brave (مفتاح خاطئ/حصة منتهية) → استمر بالخطة البديلة أدناه
        # لا يوجد مفتاح Brave أو فشل الطلب → استخدم مصادر API المجانية الأخرى
        return await _fallback_search_sources(query)

    res = await run_cmd(cmd, timeout)
    if res["returncode"] == 0 and not _result_empty(res["stdout"]):
        return res
    is_search = "curl" in cmd and any(x in cmd for x in ["search", "duckduck", "bing", "google", "wikipedia", "reddit"])
    if not is_search:
        return res
    m = re.search(r"[?&]q=([^&'\"\s]+)", cmd)
    query = urllib.parse.unquote(m.group(1).replace("+", " ")).strip() if m else ""
    if len(query) < 3:
        return res
    return await _fallback_search_sources(query, prior_result=res)


async def _fallback_search_sources(query: str, prior_result: dict = None) -> dict:
    """يجرب Brave أولاً (لو متاح)، ثم مصادر API المجانية الأخرى (DuckDuckGo
    Instant Answer, Wikipedia, HackerNews, Reddit) — كل هذي APIs رسمية
    ومصرَّح باستخدامها آلياً، لا تحتاج فتح متصفح ولا تواجه كابتشا إطلاقاً."""
    if BRAVE_API_KEY:
        res = await brave_search(query)
        if res["returncode"] == 0 and not _result_empty(res["stdout"]):
            return res

    results = []
    for src in _search_sources(query):
        r2 = await run_cmd(src["cmd"], 20)
        if not _result_empty(r2["stdout"]):
            results.append(f"[{src['name']}]\n{r2['stdout']}")
            if len(results) >= 2:
                break
    if results:
        return {"stdout": "\n\n".join(results)[:15000], "stderr": "", "returncode": 0}
    return prior_result or {"stdout": "", "stderr": "لم يُعثر على نتائج من أي مصدر", "returncode": -1}


# ════════════════════════════════════════════════════════════════
# ── FastAPI App ───────────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

app = FastAPI(title="Z-Computer-Mode v20 — Headless-Native")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True,
                   allow_methods=["*"], allow_headers=["*"])


@app.get("/", response_class=HTMLResponse)
async def root():
    with _sessions_lock:
        n = len(_sessions)
        rows = "".join(
            f"<tr><td>{uid[:30]}</td><td>{s['port']}</td>"
            f"<td>{'🟢 متصل' if s.get('active_ws') else '⚫ غير متصل'}</td></tr>"
            for uid, s in _sessions.items()
        )
    return f"""<!DOCTYPE html><html><head><meta charset="utf-8">
<title>Z Computer Mode v20</title>
<style>body{{background:#080808;color:#00d4aa;font-family:monospace;padding:40px}}
h1{{color:#4a90d9}}.ok{{color:#00ff88}}.info{{color:#888;font-size:13px}}
table{{border-collapse:collapse;width:100%}}td,th{{border:1px solid #333;padding:6px 10px;font-size:13px}}
th{{background:#1a1a2e;color:#4a90d9}}</style></head><body>
<h1>🖥️ THE Z AI — Computer Mode Server v20 (Headless-Native)</h1>
<p class="ok">✅ يعمل — بدون Xvfb، Chromium headless=new فقط</p>
<p>🌐 المتصفح المعتمد: <b>{BROWSER_BIN}</b></p>
<p>🔍 Brave Search API: <b>{'✅ مفعّل' if BRAVE_API_KEY else '❌ غير مُعرَّف (عيّن BRAVE_API_KEY)'}</b></p>
<p>🦊 Camoufox (مضاد للكشف): <b>{'✅ متاح — إرسال action: switch_engine للتفعيل' if _CAMOUFOX_LIB_AVAILABLE and CAMOUFOX_ENABLED else '❌ غير متاح'}</b></p>
<p>👥 الجلسات النشطة: <b>{n}</b> / {MAX_SESSIONS}</p>
<hr style="border-color:#333"><h3>👤 الجلسات</h3>
<table><tr><th>User ID</th><th>CDP Port</th><th>الحالة</th></tr>
{rows or '<tr><td colspan="3" style="text-align:center;color:#666">لا توجد جلسات</td></tr>'}
</table>
<hr style="border-color:#333">
<p class="info">Endpoints: /health · /ws?user_id=email (WebSocket)</p>
</body></html>"""


# ════════════════════════════════════════════════════════════════
# ── معالج الأوامر — بروتوكول مطابق 100% للواجهة الأمامية ────────
# ════════════════════════════════════════════════════════════════
# الأوامر المدعومة (كل ما ترسله index_plus_prompt_enhanced_v2-3.html فعلياً):
#   screenshot, terminal, mouse_move, mouse_click, mouse_drag, keyboard_type,
#   keyboard_hotkey, keyboard_press, scroll, clipboard_write, clipboard_read,
#   paste, open_app, open_tab, close_tab, browser_back, browser_forward,
#   browser_search, screen_info, reset_computer

# عتبة أبسط بكثير من v13: بما أن لا يوجد X server منفصل يمكن أن "يتجمد" دون
# أن يتوقف Chromium نفسه، أي فشل التقاط الآن يعني عملياً أن Chromium توقف أو
# اتصال CDP انقطع — فشلة واحدة كافية للتأكد ومحاولة إعادة الاتصال مباشرة قبل
# اللجوء لإعادة تشغيل العملية بالكامل.
_RECONNECT_THRESHOLD = 1
_FULL_RESET_THRESHOLD = 3
_MAX_ROUNDS = 4


async def handle_action(ws: WebSocket, msg: dict, sess: dict):
    action = msg.get("action", "")
    data = msg.get("data", {})

    async def send(obj):
        try:
            await ws.send_text(json.dumps(obj, ensure_ascii=False))
        except Exception:
            pass

    async def do_screenshot(label: str = "", auto: bool = False):
        """
        يلتقط ويرسل لقطة شاشة، مع تعافي تلقائي بسيط:
          فشلة 1  → أعد الاتصال بـ CDP فقط (Chromium غالباً حي، الاتصال فقط سقط)
          فشلة 3  → أعد تشغيل Chromium بالكامل (reset_session)
          بعد ذلك → أرسل computer_reset صراحة للعميل بدل الصمت
        """
        for _round in range(_MAX_ROUNDS):
            ok_browser = await ensure_browser_running(sess)
            if not ok_browser:
                sess["consecutive_failures"] += 1
            else:
                async with sess["shot_lock"]:
                    result = await capture_full(sess, quality=75)
                if result.get("data") and not result.get("is_placeholder"):
                    sess["consecutive_failures"] = 0
                    fh = _frame_hash(result["data"])
                    if not auto or fh != sess.get("last_frame_hash", ""):
                        sess["last_frame_hash"] = fh
                        await send({
                            "type": "screenshot",
                            "data": result["data"],
                            "data_grid": result.get("data_grid", ""),
                            "data_grid2": result.get("data_grid2", ""),
                            "data_grid3": result.get("data_grid3", ""),
                            "ts": int(time.time() * 1000),
                            "auto": auto, "label": label,
                            "screen_width": result["width"], "screen_height": result["height"],
                            "mouse_x": result["mouse_x"], "mouse_y": result["mouse_y"],
                            "has_grid": True,
                        })
                    return
                sess["consecutive_failures"] += 1

            fails = sess["consecutive_failures"]
            if fails >= _FULL_RESET_THRESHOLD:
                await send({"type": "computer_reset",
                           "msg": "🔄 إعادة تشغيل تلقائية للكمبيوتر بعد فشل متكرر — لحظات ويعود طبيعياً",
                           "auto": True})
                await reset_session(sess["user_id"])
                await asyncio.sleep(1.0)
                continue
            elif fails >= _RECONNECT_THRESHOLD:
                # أعد الاتصال بـ CDP فقط (أخف بكثير من إعادة تشغيل Chromium بالكامل)
                sess["cdp"] = None
                await asyncio.sleep(0.3)
                continue
            await asyncio.sleep(0.3)

        # استنفدنا كل المحاولات — أعلم العميل صراحة بدل الصمت الأبدي
        await send({"type": "computer_reset",
                   "msg": "⚠️ استمرت مشكلة الالتقاط — الرجاء طلب لقطة شاشة جديدة",
                   "auto": True})

    def bg_shot(label: str = "", delay: float = 0.5):
        async def _run():
            await asyncio.sleep(delay)
            await do_screenshot(label, auto=True)
        asyncio.create_task(_run())

    # ── reset_computer ───────────────────────────────────────
    if action == "reset_computer":
        await send({"type": "ack", "action": "reset_computer", "status": "resetting"})
        await reset_session(sess["user_id"])
        await asyncio.sleep(1.0)
        await do_screenshot("after reset")
        await send({"type": "computer_reset", "msg": "✅ تم إعادة ضبط الكمبيوتر — الشاشة جديدة تماماً"})
        return

    # ── screenshot ────────────────────────────────────────────
    if action == "screenshot":
        _task = asyncio.create_task(do_screenshot("explicit screenshot"))
        def _on_done(t: asyncio.Task):
            exc = t.exception() if not t.cancelled() else None
            if exc:
                print(f"[screenshot-task:{sess['user_id']}] ❌ {exc}")
        _task.add_done_callback(_on_done)
        return

    # ── terminal ─────────────────────────────────────────────
    if action == "terminal":
        cmd = data.get("cmd", "")
        if not cmd:
            await send({"type": "terminal_result", "stdout": "", "stderr": "no cmd", "returncode": -1})
            return
        res = await run_cmd_smart(cmd, int(data.get("timeout", 60)))
        await send({"type": "terminal_result", "cmd": cmd, "stdout": res["stdout"],
                   "stderr": res.get("stderr", ""), "returncode": res["returncode"]})
        return

    # ── يحتاج المتصفح شغالاً؟ تأكد أولاً لبقية الأوامر ──
    engine = sess.get("engine", "chromium")
    if engine == "chromium":
        await ensure_browser_running(sess)
    else:
        cam = await get_camoufox_session(sess["user_id"])
        if not cam:
            # فشل بدء Camoufox (مكتبة غير مثبَّتة/فشل إقلاع) → ارجع تلقائياً لـ Chromium
            print(f"[engine] ⚠️ فشل Camoufox لـ '{sess['user_id']}' — رجوع تلقائي لـ Chromium")
            sess["engine"] = "chromium"
            engine = "chromium"
            await ensure_browser_running(sess)

    # ── switch_engine — تبديل محرك المتصفح للمستخدم (chromium ⇄ camoufox) ──
    if action == "switch_engine":
        target = data.get("engine", "chromium")
        if target not in ("chromium", "camoufox"):
            await send({"type": "error", "msg": f"Unknown engine: '{target}'"})
            return
        if target == "camoufox" and not _CAMOUFOX_LIB_AVAILABLE:
            await send({"type": "error", "msg": "Camoufox غير مثبَّت على هذا السيرفر"})
            return
        sess["engine"] = target
        sess["last_frame_hash"] = ""
        await send({"type": "ack", "action": "switch_engine", "engine": target,
                   "msg": f"✅ تم التبديل لمحرك '{target}'"
                          + (" — متصفح مضاد للكشف، أبطأ قليلاً لكن يتجنب كابتشا المواقع المحمية بشدة"
                             if target == "camoufox" else " — المحرك الافتراضي السريع")})
        bg_shot(f"after switch to {target}", delay=0.8)
        return

    if action == "mouse_move":
        x, y = int(data.get("x", 0)), int(data.get("y", 0))
        if engine == "camoufox":
            await cam.mouse_move(x, y)
        else:
            await cdp_mouse_move(sess, x, y)
        await send({"type": "ack", "action": "mouse_move", "x": x, "y": y})

    elif action == "mouse_click":
        x, y = int(data.get("x", 0)), int(data.get("y", 0))
        btn = {"left": "left", "middle": "middle", "right": "right"}.get(data.get("button", "left"), "left")
        double = bool(data.get("double", False))
        if engine == "camoufox":
            await cam.mouse_click(x, y, btn, double)
        else:
            await cdp_mouse_click(sess, x, y, btn, double)
        await send({"type": "ack", "action": "mouse_click", "x": x, "y": y, "button": btn})
        bg_shot("after click", delay=0.5)

    elif action == "mouse_drag":
        x1, y1 = int(data.get("x1", 0)), int(data.get("y1", 0))
        x2, y2 = int(data.get("x2", 0)), int(data.get("y2", 0))
        if engine == "camoufox":
            await cam.mouse_drag(x1, y1, x2, y2)
        else:
            await cdp_mouse_drag(sess, x1, y1, x2, y2)
        await send({"type": "ack", "action": "mouse_drag"})
        bg_shot("after drag", delay=0.4)

    elif action == "keyboard_type":
        text = _safe_search(data.get("text", ""))
        if text:
            if engine == "camoufox":
                await cam.type_text(text)
            else:
                await cdp_type_text(sess, text)
        await send({"type": "ack", "action": "keyboard_type", "method": engine})

    elif action == "keyboard_hotkey":
        keys = data.get("keys", [])
        if keys:
            if engine == "camoufox":
                await cam.hotkey(keys)
            else:
                await cdp_hotkey(sess, keys)
        await send({"type": "ack", "action": "keyboard_hotkey", "keys": keys})

    elif action == "keyboard_press":
        key = data.get("key", "")
        if key:
            if engine == "camoufox":
                await cam.key_press(key)
            else:
                await cdp_key_press(sess, key)
        await send({"type": "ack", "action": "keyboard_press"})

    elif action == "scroll":
        x, y = int(data.get("x", SCREEN_W // 2)), int(data.get("y", SCREEN_H // 2))
        clicks = max(-5, min(5, int(data.get("clicks", 3))))
        if engine == "camoufox":
            await cam.scroll(x, y, delta_y=-clicks * 100)
        else:
            await cdp_scroll(sess, x, y, delta_y=-clicks * 100)
        await send({"type": "ack", "action": "scroll", "clicks": clicks})

    elif action == "clipboard_write":
        text = data.get("text", "")
        if engine != "camoufox":
            await cdp_clipboard_write(sess, text)
        # في Camoufox، النسخ الفعلي يُحاكى عبر paste مباشرة (لا حاجة لتخزين منفصل)
        await send({"type": "ack", "action": "clipboard_write", "length": len(text)})

    elif action == "clipboard_read":
        text = await cdp_clipboard_read(sess) if engine != "camoufox" else ""
        await send({"type": "clipboard_content", "text": text})

    elif action == "paste":
        text = data.get("text", "")
        if text:
            if engine == "camoufox":
                await cam.type_text(text)
            else:
                await cdp_clipboard_write(sess, text)
                await cdp_type_text(sess, text)
        await send({"type": "ack", "action": "paste"})

    elif action == "open_app":
        # في هذه البنية Chromium/Camoufox هو "التطبيق" الوحيد؛ أي أمر يحوي
        # firefox/chromium/chrome يُفهم كطلب فتح صفحة فارغة جديدة في نفس المتصفح.
        cmd = _safe_search(data.get("cmd", ""))
        url_match = re.search(r"https?://\S+", cmd)
        target_url = url_match.group(0) if url_match else "about:blank"
        if engine == "camoufox":
            await cam.navigate(target_url)
        else:
            await cdp_navigate(sess, target_url)
        await send({"type": "ack", "action": "open_app", "cmd": cmd})
        bg_shot("after open_app", delay=1.0)

    elif action == "open_tab":
        url = _safe_search(data.get("url", "") or "about:blank")
        if engine == "camoufox":
            await cam.navigate(url)
        else:
            await cdp_navigate(sess, url)
        await send({"type": "ack", "action": "open_tab", "url": url})
        bg_shot("after open_tab", delay=1.2)

    elif action == "close_tab":
        if engine == "camoufox":
            await cam.navigate("about:blank")
        else:
            await cdp_navigate(sess, "about:blank")
        await send({"type": "ack", "action": "close_tab"})
        bg_shot("after close_tab", delay=0.4)

    elif action == "browser_back":
        if engine == "camoufox":
            await cam.go_back()
        else:
            cdp = sess.get("cdp")
            if cdp and cdp.alive:
                hist = await cdp.send("Page.getNavigationHistory", timeout=4.0)
                if hist:
                    entries = hist.get("result", {}).get("entries", [])
                    idx = hist.get("result", {}).get("currentIndex", 0)
                    if idx > 0:
                        await cdp.send("Page.navigateToHistoryEntry", {"entryId": entries[idx - 1]["id"]}, timeout=5.0)
        await send({"type": "ack", "action": "browser_back"})
        bg_shot("after back", delay=0.6)

    elif action == "browser_forward":
        if engine == "camoufox":
            await cam.go_forward()
        else:
            cdp = sess.get("cdp")
            if cdp and cdp.alive:
                hist = await cdp.send("Page.getNavigationHistory", timeout=4.0)
                if hist:
                    entries = hist.get("result", {}).get("entries", [])
                    idx = hist.get("result", {}).get("currentIndex", 0)
                    if idx < len(entries) - 1:
                        await cdp.send("Page.navigateToHistoryEntry", {"entryId": entries[idx + 1]["id"]}, timeout=5.0)
        await send({"type": "ack", "action": "browser_forward"})
        bg_shot("after forward", delay=0.6)

    elif action == "browser_search":
        url = _safe_search(data.get("url", "") or data.get("query", ""))
        if not re.match(r"^https?://", url):
            url = "https://duckduckgo.com/?q=" + urllib.parse.quote_plus(url)
        if engine == "camoufox":
            await cam.navigate(url)
        else:
            await cdp_navigate(sess, url)
        await send({"type": "ack", "action": "browser_search"})
        bg_shot("after search", delay=1.0)

    elif action == "screen_info":
        info = await cdp_get_screen_info(sess)
        await send({"type": "screen_info", "width": info["width"], "height": info["height"],
                   "mouse_x": info["mouse_x"], "mouse_y": info["mouse_y"],
                   "browser": BROWSER_BIN if engine == "chromium" else "camoufox (Firefox anti-detect)",
                   "engine": engine})

    else:
        await send({"type": "error", "msg": f"Unknown action: '{action}'"})


# ════════════════════════════════════════════════════════════════
# ── WebSocket Endpoint ────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket, user_id: str = Query(default="anonymous")):
    await ws.accept()
    sess = await get_or_create_session(user_id, ws)
    await ensure_browser_running(sess)

    async def _heartbeat():
        while True:
            await asyncio.sleep(20)
            try:
                await ws.send_text(json.dumps({"type": "ping", "ts": int(time.time() * 1000)}))
            except Exception:
                break

    hb_task = asyncio.create_task(_heartbeat())

    try:
        info = await cdp_get_screen_info(sess)
        await ws.send_text(json.dumps({
            "type": "connected",
            "screen_width": info["width"], "screen_height": info["height"],
            "browser": BROWSER_BIN, "display": f"headless:{sess['port']}",
            "user_id": user_id, "session_id": id(ws),
            "msg": f"Z Computer Mode v20 | User: {user_id} | Port: {sess['port']} | "
                   f"Browser: {BROWSER_BIN} (headless=new) | Screen: {info['width']}x{info['height']}",
        }, ensure_ascii=False))

        # لقطة شاشة أولية
        result = await capture_full(sess, quality=72)
        if result.get("data"):
            await ws.send_text(json.dumps({
                "type": "screenshot", "data": result["data"],
                "data_grid": result.get("data_grid", ""), "data_grid2": result.get("data_grid2", ""),
                "data_grid3": result.get("data_grid3", ""), "ts": int(time.time() * 1000),
                "label": "Initial screen", "screen_width": result["width"], "screen_height": result["height"],
                "mouse_x": result["mouse_x"], "mouse_y": result["mouse_y"], "has_grid": True,
            }, ensure_ascii=False))
            sess["last_frame_hash"] = _frame_hash(result["data"])
    except Exception as e:
        print(f"[ws:{user_id}] خطأ عند البدء: {e}")

    try:
        while True:
            raw = await ws.receive_text()
            try:
                msg = json.loads(raw)
                if msg.get("type") == "pong":
                    continue
                await handle_action(ws, msg, sess)
            except json.JSONDecodeError:
                pass
    except WebSocketDisconnect:
        pass
    except Exception as e:
        print(f"[ws:{user_id}] خطأ: {e}")
    finally:
        hb_task.cancel()
        await destroy_session_ws(user_id, ws)


# ════════════════════════════════════════════════════════════════
# ── REST Endpoints ────────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

@app.get("/screenshot")
async def rest_screenshot(user_id: str = "anonymous"):
    with _sessions_lock:
        sess = _sessions.get(user_id)
    if not sess:
        return JSONResponse({"error": "no session for this user_id"}, status_code=404)
    await ensure_browser_running(sess)
    result = await capture_full(sess, quality=75)
    return JSONResponse({
        "image": result["data"], "image_grid": result.get("data_grid", ""),
        "image_grid2": result.get("data_grid2", ""), "image_grid3": result.get("data_grid3", ""),
        "ts": int(time.time() * 1000), "screen_width": result["width"], "screen_height": result["height"],
        "mouse_x": result["mouse_x"], "mouse_y": result["mouse_y"], "has_grid": True,
        "user_id": user_id,
    })


@app.post("/terminal")
async def rest_terminal(body: dict):
    return JSONResponse(await run_cmd_smart(body.get("cmd", ""), body.get("timeout", 60)))


@app.get("/health")
async def health():
    with _sessions_lock:
        n = len(_sessions)
        users = [
            {"user_id": uid, "cdp_port": s["port"], "connected": bool(s.get("active_ws")),
             "browser_alive": bool(s.get("cdp") and s["cdp"].alive),
             "engine": s.get("engine", "chromium")}
            for uid, s in _sessions.items()
        ]
    return {"status": "ok", "version": "v20.1-headless-native+camoufox", "browser": BROWSER_BIN,
           "brave_search_enabled": bool(BRAVE_API_KEY),
           "camoufox_available": bool(_CAMOUFOX_LIB_AVAILABLE and CAMOUFOX_ENABLED),
           "active_sessions": n, "max_sessions": MAX_SESSIONS, "users": users}


# ════════════════════════════════════════════════════════════════
# ── Background Cleanup ────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

async def _cleanup_idle_sessions():
    """يُزيل جلسات المستخدمين غير النشطين (بدون اتصال WebSocket منذ أكثر من
    ساعة) — يوقف Chromium الخاص بهم فعلياً لتحرير الذاكرة."""
    while True:
        await asyncio.sleep(1800)
        now = time.time()
        with _sessions_lock:
            idle_uids = [uid for uid, s in _sessions.items()
                        if not s.get("active_ws") and now - s.get("created", now) > 3600]
        for uid in idle_uids:
            print(f"[cleanup] 🗑️ إزالة جلسة خاملة '{uid}'")
            await destroy_session_full(uid)


@app.on_event("startup")
async def startup():
    asyncio.create_task(_cleanup_idle_sessions())
    print("✅ Z Computer Mode v20 جاهز — Headless-Native (بدون Xvfb)")
    print(f"   المتصفح المعتمد: {BROWSER_BIN} (--headless=new)")
    print(f"   مرشحو المتصفح المفحوصون: {', '.join(_BROWSER_CANDIDATES)}")
    print(f"   أقصى جلسات متزامنة: {MAX_SESSIONS}")
    print(f"   نطاق منافذ CDP: {CDP_PORT_BASE}+")
    print(f"   websockets lib: {'✅' if _WS_LIB_AVAILABLE else '❌ (pip install websockets)'}")
    if BRAVE_API_KEY:
        print("   🔍 Brave Search API: ✅ مفعّل — البحث سيتم عبره أولاً (بدون كابتشا)")
    else:
        print("   🔍 Brave Search API: ❌ غير مُعرَّف — عيّن BRAVE_API_KEY لتفعيله")
        print("      (بدونه، البحث سيعتمد على DuckDuckGo/Wikipedia APIs المجانية كبديل)")
    if _CAMOUFOX_LIB_AVAILABLE and CAMOUFOX_ENABLED:
        print("   🦊 Camoufox (متصفح مضاد للكشف): ✅ متاح — عبر action: switch_engine {engine: camoufox}")
    else:
        print("   🦊 Camoufox: ❌ غير متاح — pip install camoufox playwright && playwright install firefox")
    print("   Connect: wss://your-space.hf.space/ws?user_id=EMAIL")


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 7860))
    uvicorn.run("app:app", host="0.0.0.0", port=port, log_level="info")