app.py

"""
THE Z AI — Computer Mode Server v11 — PER-USER ISOLATED DISPLAYS
=================================================================
كل مستخدم لديه:
  ✅ Display Xvfb خاص (:100, :101, :102 ...)
  ✅ متصفح Firefox معزول تماماً لا يتشارك مع أي مستخدم آخر
  ✅ لقطات شاشة مستقلة — ما يراه مستخدم A لا يختلط بمستخدم B
  ✅ تاريخ terminal مستقل لكل مستخدم
  ✅ عند كل اتصال جديد للمستخدم: نفس الـ display (لا يُعاد إنشاء Xvfb)
  ✅ عند طلب ريستارت / كمبيوتر جديد: Xvfb + Firefox يُعادان تماماً

الفكرة:
  - كل user_id (email) ↦ display رقم ثابت (مثلاً :100 للأول، :101 للثاني)
  - إذا المستخدم يتصل مجدداً، نُعيد استخدام نفس الـ display الخاص به
  - إذا طلب ريستارت، نقتل الـ Xvfb والمتصفح ونُعيد إنشاءهما
  - إذا المستخدم جديد، نخصص له display جديد ونُشغّل Xvfb له
"""

import asyncio
import base64
import hashlib
import io
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.parse
import threading

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

# ════════════════════════════════════════════════════════════════
# ── إعداد Display Pool ─────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

# بداية نطاق الـ displays الافتراضية (يمكن تغييره)
DISPLAY_BASE = 100
DISPLAY_MAX  = 200   # أقصى عدد مستخدمين متزامنين

_display_lock = threading.Lock()

# user_id → display_info
# display_info = {
#   "display": ":101",
#   "xvfb_proc": subprocess.Popen | None,
#   "browser_proc": subprocess.Popen | None,
#   "last_bg_shot_ts": float,
#   "last_bg_hash": str,
#   "active_ws": WebSocket | None,   # الاتصال الحالي لهذا المستخدم
# }
_user_displays: dict[str, dict] = {}
_display_numbers: set[int] = set()  # أرقام الـ displays المستخدمة

# semaphore للتحكم في عمليات terminal (global لكل السيرفر)
_terminal_sem = asyncio.Semaphore(8)

# ════════════════════════════════════════════════════════════════
# ── اكتشاف المتصفح ────────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

def _detect_browser() -> str:
    for c in ["firefox", "firefox-esr", "chromium-browser", "chromium", "google-chrome"]:
        r = subprocess.run(["which", c], capture_output=True, text=True)
        if r.returncode == 0 and r.stdout.strip():
            return c
    return "firefox"

BROWSER = _detect_browser()
print(f"🌐 Browser: {BROWSER}")

# ── CLI flags ثابتة تُضاف لكل تشغيل Firefox ──
# تسرّع الإقلاع (لا GPU فيزيائي حقيقي في Xvfb فلا داعي لمحاولة
# التسريع المرئي الذي يفشل دائماً هنا ويسبّب تأخيراً في الإقلاع)،
# وتمنع أي nag screens أو crash-reporter windows من الظهور.
FIREFOX_CLI_FLAGS = ["--no-remote", "--new-instance"]
FIREFOX_ENV_EXTRA = {
    # يمنع محاولات تسريع GPU غير المتوفرة في Xvfb من إبطاء الإقلاع
    "MOZ_DISABLE_GPU_SANDBOX": "1",
    "MOZ_ACCELERATED": "0",
}

# ════════════════════════════════════════════════════════════════
# ── Firefox Profile (per-user, isolated, crash-recovery DISABLED) ─
# ════════════════════════════════════════════════════════════════
# المشكلة الجذرية التي هذا القسم يحلّها:
# عند قتل Firefox (حتى بـ pkill "لطيف")، لا يُغلق بشكل نظيف دائماً على
# سيرفر بموارد محدودة/Xvfb، فتُسجَّل الجلسة على أنها "crashed" داخل
# sessionstore.jsonlz4 الخاص بالبروفايل. في المرة التالية يفتح Firefox
# ويعرض شاشة "Sorry. We're having trouble getting your pages back" بدل
# الصفحة المطلوبة — وهذه هي الشاشة السوداء/الفاشلة التي تصل كلقطة "ناجحة"
# تقنياً (ليست سوداء فعلاً) لكنها لا تحتوي على المحتوى المطلوب أبداً.
# الحل الجذري: بروفايل خاص بكل مستخدم (isolated) مع user.js يعطّل تماماً:
#   - استرجاع الجلسة بعد التحطّم (session restore prompt)
#   - أي محاولة لإعادة فتح تبويبات سابقة
# بهذا، كل فتح لفايرفوكس يبدأ صفحة نظيفة فارغة دائماً، بغض النظر عن كيفية
# إغلاقه سابقاً.

FIREFOX_PROFILE_PREFS = """
// ── تعطيل استرجاع الجلسة نهائياً (سبب شاشة "Restore Session") ──
user_pref("browser.sessionstore.resume_from_crash", false);
user_pref("browser.sessionstore.resume_session_once", false);
user_pref("browser.sessionstore.max_resumed_crashes", -1);
user_pref("browser.sessionstore.restore_on_demand", false);
user_pref("browser.sessionstore.enabled", false);
user_pref("browser.sessionstore.privacy_level", 2);
user_pref("browser.startup.page", 0);
user_pref("browser.startup.homepage_override.mstone", "ignore");
user_pref("toolkit.startup.max_resumed_crashes", -1);
// تعطيل نافذة "Restore previous session" ونوافذ الأزمات المختلفة
user_pref("browser.sessionstore.max_tabs_undo", 0);
user_pref("browser.sessionstore.max_windows_undo", 0);
user_pref("browser.tabs.crashReporting.sendReport", false);
user_pref("browser.crashReports.unsubmittedCheck.autoSubmit2", false);
user_pref("browser.crashReports.unsubmittedCheck.enabled", false);
user_pref("toolkit.crashreporter.infoURL", "");
// تعطيل شاشة "Restore" وكذلك أي إشعار تحطّم
user_pref("browser.sessionstore.resumeFromCrash", false);
user_pref("dom.ipc.plugins.flash.subprocess.crashreporter.enabled", false);
// إلغاء استعادة آخر جلسة تماماً + دائماً about:blank
user_pref("browser.startup.homepage", "about:blank");
// تعطيل تحديثات ورسائل onboarding التي قد تبطئ أو تحجب اللقطة
user_pref("browser.shell.checkDefaultBrowser", false);
user_pref("browser.aboutwelcome.enabled", false);
user_pref("browser.startup.firstrunSkipsHomepage", true);
user_pref("startup.homepage_welcome_url", "");
user_pref("startup.homepage_welcome_url.additional", "");
user_pref("browser.uitour.enabled", false);
user_pref("browser.newtabpage.activity-stream.feeds.telemetry", false);
user_pref("datareporting.policy.dataSubmissionEnabled", false);
user_pref("app.normandy.enabled", false);
user_pref("app.update.enabled", false);
user_pref("app.update.auto", false);
// ── إصلاحات تجمد التحميل ("Connecting to ... تبقى للأبد") ──
// السبب الغالب على سيرفرات مجانية/container: محاولة IPv6 أولاً (غير
// مدعوم فعلياً على أغلب مزودي Render/HuggingFace المجانية) قبل الرجوع
// لـ IPv4 بعد timeout طويل جداً (قد يصل 20-30 ثانية لكل طلب DNS)،
// وهذا بالضبط السبب الذي يظهر كشاشة "Connecting..." معلقة لا تتقدم أبداً.
user_pref("network.dns.disableIPv6", true);
user_pref("network.http.fast-fallback-to-IPv4", true);
// تقليل مهلة connect الإجمالية لكل طلب من 15 ثانية (الافتراضي) إلى 8 ثوانٍ
// — أفضل أن يفشل الطلب بسرعة ويقرر الذكاء الاصطناعي بديلاً من أن يبقى عالقاً طويلاً.
user_pref("network.http.connection-timeout", 8);
user_pref("network.http.response.timeout", 12);
// تعطيل predictive prefetching/speculative connect التي تستهلك عرض حزمة على
// سيرفر محدود الموارد بدون فائدة حقيقية هنا.
user_pref("network.dns.disablePrefetch", true);
user_pref("network.prefetch-next", false);
user_pref("network.predictor.enabled", false);
user_pref("network.http.speculative-parallel-limit", 0);
// تعطيل safebrowsing الذي يستدعي طلبات خارجية إضافية عند فتح كل صفحة
// (يُبطئ التحميل أو يعلقه لو الطلب لقوائم Google فشل أو استغرق).
user_pref("browser.safebrowsing.malware.enabled", false);
user_pref("browser.safebrowsing.phishing.enabled", false);
user_pref("browser.safebrowsing.downloads.enabled", false);
user_pref("browser.safebrowsing.provider.google4.updateURL", "");
user_pref("browser.safebrowsing.provider.google.updateURL", "");
// تعطيل telemetry/captive-portal checks التي تفتح اتصالات خلفية غير ضرورية عند الإقلاع
user_pref("network.captive-portal-service.enabled", false);
user_pref("network.connectivity-service.enabled", false);
user_pref("toolkit.telemetry.server", "");
// تقليل عدد اتصالات HTTP المتزامنة لكل دومين — يقلل الضغط على شبكة محدودة
// النطاق للسيرفرات المجانية ويقلل احتمال التعلق.
user_pref("network.http.max-persistent-connections-per-server", 4);
user_pref("network.http.max-connections", 48);
"""

def _ensure_firefox_profile(user_id: str, display: str) -> str:
    """
    يُنشئ (أو يُعيد استخدام) بروفايل Firefox مخصص ونظيف لهذا المستخدم،
    مع إعدادات تعطّل استرجاع الجلسة (crash recovery) نهائياً.
    يُعيد المسار الكامل للبروفايل.
    """
    safe_id = re.sub(r"[^a-zA-Z0-9_.-]", "_", user_id) or "anon"
    profile_dir = os.path.expanduser(f"~/.zai_ff_profiles/{safe_id}")
    try:
        os.makedirs(profile_dir, exist_ok=True)
        prefs_path = os.path.join(profile_dir, "user.js")
        with open(prefs_path, "w", encoding="utf-8") as f:
            f.write(FIREFOX_PROFILE_PREFS)
    except Exception as e:
        print(f"[profile] ⚠️ failed to prepare profile for {user_id}: {e}")
    return profile_dir


def _wipe_firefox_session_data(profile_dir: str):
    """
    يمسح ملفات الجلسة الفاسدة (sessionstore) + أي lock متبقٍّ.
    هذا ضروري لأن مجرد تعطيل session restore في user.js لا يمسح ملفات
    جلسة سابقة موجودة بالفعل على القرص من قبل تفعيل هذه الإعدادات.
    """
    try:
        subprocess.run(
            f"rm -f '{profile_dir}'/lock '{profile_dir}'/.parentlock "
            f"'{profile_dir}'/sessionstore.jsonlz4 "
            f"'{profile_dir}'/sessionstore-backups/*.jsonlz4 "
            f"'{profile_dir}'/sessionCheckpoints.json 2>/dev/null",
            shell=True, timeout=5, capture_output=True
        )
    except Exception:
        pass


# ════════════════════════════════════════════════════════════════
# ── Xvfb Management (per-display) ────────────────────────────
# ════════════════════════════════════════════════════════════════

def _next_free_display() -> int:
    """يجد رقم display حر غير مستخدم."""
    used = _display_numbers.copy()
    for n in range(DISPLAY_BASE, DISPLAY_MAX):
        if n not in used:
            return n
    # إذا امتلأت القائمة — أعد استخدام أقدم display غير نشط
    return DISPLAY_BASE


def _start_xvfb(display: str) -> "subprocess.Popen | None":
    """يُشغّل Xvfb على display محدد ويُعيد الـ Popen أو None."""
    # تحقق أولاً: هل الـ display يعمل بالفعل؟
    try:
        r = subprocess.run(
            ["xdpyinfo", "-display", display],
            capture_output=True, timeout=3
        )
        if r.returncode == 0:
            print(f"[xvfb] ✅ Display {display} already active")
            return None  # يعمل بالفعل بدون Popen نشغّله
    except Exception:
        pass

    try:
        # ── دقة 1280x800 بدل 1920x1080 ──────────────────────────────
        # هذا يقلل حجم بيانات كل لقطة شاشة بنسبة ~60%، وبالتالي يسرّع كل
        # مراحل الالتقاط والترميز بشكل كبير على سيرفر بموارد محدودة.
        # نفس الدقة المستخدمة في التطبيقات المرجعية لـ computer-use.
        proc = subprocess.Popen(
            ["Xvfb", display, "-screen", "0", "1280x800x24",
             "-nolisten", "tcp", "-ac"],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
        )
        # انتظر حتى يصبح جاهزاً (max 6 ثواني)
        for _ in range(60):
            time.sleep(0.1)
            try:
                r = subprocess.run(
                    ["xdpyinfo", "-display", display],
                    capture_output=True, timeout=2
                )
                if r.returncode == 0:
                    print(f"[xvfb] ✅ Xvfb started on {display}")
                    # إصلاح: Xvfb بدون أي رسم لاحق يكون framebuffer فارغاً تماماً
                    # (لا لون خلفية أصلاً)، فأول لقطة شاشة قد تبدو "فارغة بشكل
                    # مربك" رغم أنها التقاط ناجح تقنياً لشاشة نظيفة فعلاً. نرسم
                    # لوناً محايداً بسيطاً فوراً عبر xsetroot لتوضيح أن الشاشة
                    # جاهزة وفارغة عمداً (سطح مكتب نظيف) لا أنها خطأ في الالتقاط.
                    try:
                        env_xs = {**os.environ, "DISPLAY": display}
                        subprocess.run(
                            ["xsetroot", "-solid", "#2b2b2b"],
                            env=env_xs, timeout=2, capture_output=True
                        )
                    except Exception:
                        pass
                    return proc
            except Exception:
                continue
        print(f"[xvfb] ⚠️ Xvfb may not be ready yet on {display}")
        return proc
    except FileNotFoundError:
        print(f"[xvfb] ❌ Xvfb binary not found")
        return None
    except Exception as e:
        print(f"[xvfb] ❌ Failed to start on {display}: {e}")
        return None


def _kill_proc(proc: "subprocess.Popen | None"):
    """يقتل process بأمان."""
    if not proc:
        return
    try:
        proc.terminate()
        proc.wait(timeout=3)
    except Exception:
        try:
            proc.kill()
        except Exception:
            pass


def _kill_display_processes(display: str):
    """يقتل كل العمليات المرتبطة بـ display معين."""
    env_d = {**os.environ, "DISPLAY": display}
    # قتل المتصفح — SIGKILL مباشر (-9) لتفادي أي حالة إغلاق نصفي تُسجَّل
    # لاحقاً كتحطّم (crash) داخل Firefox وتُنتج شاشة "Restore Session"
    for b in ["firefox", "firefox-esr", "chromium", "chrome"]:
        try:
            subprocess.run(
                ["pkill", "-9", "-f", f"[{b[0]}]{b[1:]}.*{display}"],
                timeout=3, capture_output=True
            )
        except Exception:
            pass
    # قتل Xvfb على هذا الـ display
    try:
        subprocess.run(
            ["pkill", "-f", f"[X]vfb {display}"],
            timeout=3, capture_output=True
        )
    except Exception:
        pass
    time.sleep(0.5)


# ════════════════════════════════════════════════════════════════
# ── Session Management (per user_id) ─────────────────────────
# ════════════════════════════════════════════════════════════════

async def get_or_create_user_session(user_id: str, ws: WebSocket) -> dict:
    """
    يُعيد session المستخدم (أو يُنشئ واحدة جديدة إذا لم تكن موجودة).
    كل user_id ← display خاص + Xvfb خاص.
    """
    with _display_lock:
        if user_id in _user_displays:
            sess = _user_displays[user_id]
            # حدّث الـ WebSocket الحالي
            sess["active_ws"] = ws
            print(f"[session] 🔁 Reconnected user '{user_id}' on display {sess['display']}")
            return sess

        # مستخدم جديد — خصص له display
        disp_num = _next_free_display()
        _display_numbers.add(disp_num)
        display = f":{disp_num}"

        sess = {
            "user_id": user_id,
            "display": display,
            "xvfb_proc": None,   # سيُشغَّل لاحقاً
            "browser_proc": None,
            "last_bg_shot_ts": 0.0,
            "last_bg_hash": "",
            "active_ws": ws,
            "created": time.time(),
        }
        _user_displays[user_id] = sess
        print(f"[session] ✅ New user '{user_id}' → display {display}")
        return sess


async def reset_user_computer(user_id: str) -> dict:
    """
    يُعيد ضبط الكمبيوتر الافتراضي للمستخدم:
    يقتل Xvfb والمتصفح ويُشغّل Xvfb جديداً نظيفاً.
    """
    with _display_lock:
        sess = _user_displays.get(user_id)
        if not sess:
            return {}

        display = sess["display"]
        print(f"[reset] 🔄 Resetting computer for user '{user_id}' on {display}")

        # قتل المتصفح — SIGKILL مباشر (بدل terminate اللطيف) لأن الأخير قد
        # يترك فايرفوكس في حالة نصف-مغلقة تُسجَّل كتحطّم (crash) في المرة
        # القادمة، وهذا بالضبط ما يُنتج شاشة "Restore Session" لاحقاً.
        browser_proc = sess.get("browser_proc")
        if browser_proc:
            try:
                browser_proc.kill()
                browser_proc.wait(timeout=3)
            except Exception:
                pass
        sess["browser_proc"] = None

        # قتل Xvfb
        _kill_proc(sess.get("xvfb_proc"))
        sess["xvfb_proc"] = None

        # تنظيف شامل لهذا الـ display
        _kill_display_processes(display)

        # ── تنظيف شامل لبيانات جلسة Firefox الفاسدة (السبب الجذري لشاشة
        # "Restore Session") — ليس فقط lock files بل sessionstore كاملاً،
        # في كل من البروفايل الافتراضي وأي بروفايل مخصص لهذا المستخدم ──
        try:
            subprocess.run(
                "rm -f ~/.mozilla/firefox/*/lock ~/.mozilla/firefox/*/.parentlock "
                "~/.mozilla/firefox/*/sessionstore.jsonlz4 "
                "~/.mozilla/firefox/*/sessionstore-backups/*.jsonlz4 "
                "~/.mozilla/firefox/*/sessionCheckpoints.json 2>/dev/null",
                shell=True, timeout=5, capture_output=True
            )
        except Exception:
            pass
        safe_id = re.sub(r"[^a-zA-Z0-9_.-]", "_", user_id) or "anon"
        custom_profile = os.path.expanduser(f"~/.zai_ff_profiles/{safe_id}")
        _wipe_firefox_session_data(custom_profile)

    # تشغيل Xvfb جديد (خارج الـ lock لأن start_xvfb يستغرق وقتاً)
    new_proc = await asyncio.to_thread(_start_xvfb, display)
    with _display_lock:
        if user_id in _user_displays:
            _user_displays[user_id]["xvfb_proc"] = new_proc
            _user_displays[user_id]["last_bg_shot_ts"] = 0.0
            _user_displays[user_id]["last_bg_hash"] = ""

    print(f"[reset] ✅ Computer reset done for '{user_id}' on {display}")
    return _user_displays.get(user_id, {})


async def ensure_xvfb_for_session(sess: dict):
    """يتأكد أن Xvfb يعمل لهذا الـ session — يُشغّله إذا لم يكن كذلك."""
    display = sess["display"]

    # تحقق إذا كان يعمل بالفعل
    try:
        r = subprocess.run(
            ["xdpyinfo", "-display", display],
            capture_output=True, timeout=3
        )
        if r.returncode == 0:
            return  # يعمل
    except Exception:
        pass

    # شغّله
    proc = await asyncio.to_thread(_start_xvfb, display)
    with _display_lock:
        if sess["user_id"] in _user_displays:
            _user_displays[sess["user_id"]]["xvfb_proc"] = proc


async def destroy_user_ws(user_id: str, ws: WebSocket):
    """
    يُزيل الـ WebSocket من الـ session عند انقطاع الاتصال.
    لا يحذف الـ session نفسها — المستخدم يحتفظ بكمبيوتره.
    """
    with _display_lock:
        sess = _user_displays.get(user_id)
        if sess and sess.get("active_ws") is ws:
            sess["active_ws"] = None
    print(f"[session] 📴 User '{user_id}' disconnected (session kept)")


# ════════════════════════════════════════════════════════════════
# ── Screenshot Engine (per-display) ──────────────────────────
# ════════════════════════════════════════════════════════════════

def _is_black_screen(img) -> bool:
    try:
        small = img.resize((100, 100))
        pixels = list(small.getdata())
        avg = sum(sum(p[:3]) for p in pixels) / (len(pixels) * 3 * 255)
        return avg < 0.04
    except Exception:
        return False


def _load_capture_image(path: str):
    from PIL import Image
    if not path or not os.path.exists(path):
        return None, 0, 0
    if os.path.getsize(path) < 1024:
        return None, 0, 0
    try:
        img = Image.open(path).convert("RGB")
        w, h = img.size
        if w < 100 or h < 100:
            return None, 0, 0
        return img, w, h
    except Exception as ex:
        print(f"[cap] load error for {path}: {ex}")
        return None, 0, 0


def _capture_via_import_pipe(display: str):
    """
    الطريقة الأساسية والوحيدة: `import -window root` مع الإخراج مباشرة
    عبر stdout (pipe) — بدون أي ملف مؤقت على القرص. أسرع بشكل ملحوظ من
    الكتابة لملف ثم إعادة قراءته، خصوصاً على تخزين شبكي بطيء.
    """
    from PIL import Image
    env = {**os.environ, "DISPLAY": display}
    try:
        r = subprocess.run(
            ["import", "-window", "root", "-silent", "png:-"],
            env=env, timeout=6, capture_output=True
        )
        if r.returncode != 0 or not r.stdout or len(r.stdout) < 500:
            return None, 0, 0
        img = Image.open(io.BytesIO(r.stdout)).convert("RGB")
        w, h = img.size
        if w < 100 or h < 100:
            return None, 0, 0
        return img, w, h
    except FileNotFoundError:
        print(f"[cap:{display}] import (ImageMagick) not installed")
        return None, 0, 0
    except Exception as e:
        print(f"[cap:{display}] import-pipe: {e}")
        return None, 0, 0


def _capture_via_xlib_direct(display: str):
    """
    خط الدفاع الثاني والأخير: قراءة X11 framebuffer مباشرة عبر
    python-xlib، بدون subprocess إطلاقاً (أسرع من أي أداة خارجية،
    ويعمل حتى لو ImageMagick غير مثبّت على السيرفر).
    """
    from PIL import Image
    try:
        from Xlib import display as Xdisp, X
        xd = Xdisp.Display(display)
        root = xd.screen().root
        geom = root.get_geometry()
        w, h = geom.width, geom.height
        raw = root.get_image(0, 0, w, h, X.ZPixmap, 0xFFFFFFFF)
        img = Image.frombuffer("RGB", (w, h), raw.data, "raw", "BGRX", 0, 1)
        xd.close()
        return img, w, h
    except Exception as e:
        print(f"[cap:{display}] xlib-direct: {e}")
        return None, 0, 0


def _capture_raw(display: str) -> tuple:
    """
    يلتقط الشاشة بأسرع طريقة موثوقة: import-pipe أولاً (سريعة جداً)،
    وعند فشلها فقط xlib-direct كبديل واحد. بحد أقصى محاولتين سريعتين
    لكل طريقة (وليس 9 محاولات كالسابق) — لأن كلا الطريقتين إما تنجحان
    فوراً أو تفشلان لسبب دائم، وإعادة المحاولة أكثر من مرتين لا تُغيّر
    النتيجة، فقط تُبطئ الاستجابة.
    """
    for method_name, method in (
        ("import-pipe", _capture_via_import_pipe),
        ("xlib-direct", _capture_via_xlib_direct),
    ):
        for attempt in range(2):
            try:
                img, w, h = method(display)
                if img and not _is_black_screen(img):
                    if attempt > 0:
                        print(f"[cap:{display}] ✅ {method_name} succeeded on retry")
                    return img, w, h
            except Exception as e:
                print(f"[cap:{display}] {method_name}: {e}")
            if attempt == 0:
                time.sleep(0.3)

    print(f"[cap:{display}] ⚠️ All methods failed — placeholder")
    from PIL import Image as PILImg, ImageDraw
    sw, sh = _get_screen_size(display)
    img = PILImg.new("RGB", (sw or 1280, sh or 800), (15, 20, 40))
    draw = ImageDraw.Draw(img)
    draw.rectangle([(0, 0), (sw, 55)], fill=(30, 40, 80))
    draw.text((10, 10), f"⚠️ Screenshot failed — DISPLAY={display}", fill=(255, 120, 80))
    draw.text((10, 32), "Methods tried: import-pipe, xlib-direct", fill=(130, 130, 150))
    return img, sw or 1280, sh or 800


def _get_screen_size(display: str) -> tuple:
    try:
        r = subprocess.run(
            ["xdotool", "getdisplaygeometry"],
            env={**os.environ, "DISPLAY": display},
            capture_output=True, text=True, timeout=5
        )
        parts = r.stdout.strip().split()
        return int(parts[0]), int(parts[1])
    except Exception:
        return 1280, 800


def _get_mouse_pos(display: str) -> tuple:
    try:
        r = subprocess.run(
            ["xdotool", "getmouselocation"],
            env={**os.environ, "DISPLAY": display},
            capture_output=True, text=True, timeout=5
        )
        mx = int(re.search(r"x:(\d+)", r.stdout).group(1))
        my = int(re.search(r"y:(\d+)", r.stdout).group(1))
        return mx, my
    except Exception:
        return 0, 0


def _draw_mouse_marker(draw, msx, msy, color=(255, 50, 50, 240)):
    """يرسم علامة الماوس (دائرة + خطوط تقاطع) في نقطة محددة على صورة."""
    r = 11
    line_color = (color[0], color[1], color[2], 200)
    draw.ellipse([(msx-r, msy-r), (msx+r, msy+r)], outline=color, width=2)
    draw.line([(msx-18, msy), (msx+18, msy)], fill=line_color, width=1)
    draw.line([(msx, msy-18), (msx, msy+18)], fill=line_color, width=1)


def _render_grid_variant(base_img, ow, oh, sw, sh, mx, my, display,
                         grid_step: int, line_color: tuple, label_every: int,
                         major_only_labels: bool = True):
    """
    يرسم نسخة شبكة إحداثيات واحدة فوق نسخة من الصورة الأساسية.
      grid_step        → المسافة بالبكسل الحقيقي بين كل خط شبكة.
      line_color       → لون الخطوط والأرقام (RGBA).
      label_every      → كل كم بكسل تُكتب فيه تسمية إحداثي (x,y) كاملة عند التقاطعات.
      major_only_labels → إذا True، الأرقام على الحواف تظهر فقط عند خطوط "رئيسية".
    """
    from PIL import ImageDraw
    img = base_img.copy()
    draw = ImageDraw.Draw(img, "RGBA")

    step_x = max(1, int(grid_step * sw / ow))
    step_y = max(1, int(grid_step * sh / oh))
    minor_color = (line_color[0], line_color[1], line_color[2], 35)
    major_color = (line_color[0], line_color[1], line_color[2], 110)
    text_color = line_color

    x_sc, x_r = step_x, grid_step
    x_majors = []
    while x_sc < sw:
        is_major = (x_r % label_every == 0)
        draw.line([(x_sc, 0), (x_sc, sh)], fill=(major_color if is_major else minor_color), width=1)
        if is_major or not major_only_labels:
            draw.rectangle([(x_sc+1, 2), (x_sc+34, 15)], fill=(0, 0, 0, 175))
            draw.text((x_sc+2, 3), str(x_r), fill=text_color)
        if is_major:
            x_majors.append((x_sc, x_r))
        x_sc += step_x; x_r += grid_step

    y_sc, y_r = step_y, grid_step
    y_majors = []
    while y_sc < sh:
        is_major = (y_r % label_every == 0)
        draw.line([(0, y_sc), (sw, y_sc)], fill=(major_color if is_major else minor_color), width=1)
        if is_major or not major_only_labels:
            draw.rectangle([(2, y_sc+1), (38, y_sc+14)], fill=(0, 0, 0, 175))
            draw.text((3, y_sc+2), str(y_r), fill=text_color)
        if is_major:
            y_majors.append((y_sc, y_r))
        y_sc += step_y; y_r += grid_step

    for (xs, xr) in x_majors:
        for (ys, yr) in y_majors:
            label = f"{xr},{yr}"
            tw = 6 * len(label) + 4
            draw.rectangle([(xs+2, ys+2), (xs+2+tw, ys+13)], fill=(0, 0, 0, 150))
            draw.text((xs+4, ys+2), label, fill=text_color)

    msx = int(mx * sw / ow)
    msy = int(my * sh / oh)
    _draw_mouse_marker(draw, msx, msy, color=(line_color[0], line_color[1], line_color[2], 240))

    final = img.convert("RGB")
    draw2 = ImageDraw.Draw(final)
    draw2.rectangle([(0, 0), (sw, 20)], fill=(0, 0, 0))
    draw2.text((4, 3), f"SCREEN {ow}x{oh} | MOUSE:({mx},{my}) | GRID={grid_step}px | DSP:{display}", fill=(0, 220, 160))
    draw2.rectangle([(0, sh-20), (sw, sh)], fill=(0, 0, 0))
    draw2.text((4, sh-17), "CLICKCOORDS = numbers at every intersection (real screen pixels)", fill=(180, 180, 70))
    return final


def capture_with_grid(display: str, scale: float = 0.85, quality: int = 72,
                      force_mx: int | None = None,
                      force_my: int | None = None,
                      grid_step: int = 50) -> dict:
    """
    يلتقط الشاشة من display محدد ويُنتج 3 نسخ من نفس اللقطة بالضبط (نفس اللحظة):
      - "data"       → النسخة العادية (نظيفة تماماً) + علامة الماوس فقط، بدون أي Grid.
      - "data_grid"  → شبكة إحداثيات عادية (كل 50px)، أخضر/سماوي — للذكاء الاصطناعي.
      - "data_grid2" → شبكة إحداثيات أدق (كل 20px)، أحمر — للنقرات الدقيقة.
    —— تبسيط v13: نسختان فقط (بدل 3) لتقليل عدد عمليات resize/encode لكل
    لقطة بنسبة قرابة س4إلى س2، ودقة المقاس الواحد 1280x800 الجديدة لم
    تعد تحتاج لثلاث تدرجات دقة منفصلة لتحديد الإحداثيات.
    """
    from PIL import Image, ImageDraw

    img, ow, oh = _capture_raw(display)
    if img is None:
        return {"data": "", "data_grid": "", "data_grid2": "", "data_grid3": "",
                "width": 1280, "height": 800, "mouse_x": 0, "mouse_y": 0}

    mx, my = (force_mx, force_my) if force_mx is not None else _get_mouse_pos(display)

    sw = int(ow * scale)
    sh = int(oh * scale)
    base_img = img.resize((sw, sh), Image.LANCZOS)
    msx = int(mx * sw / ow)
    msy = int(my * sh / oh)

    clean_img = base_img.copy()
    draw_clean = ImageDraw.Draw(clean_img, "RGBA")
    _draw_mouse_marker(draw_clean, msx, msy)
    clean_final = clean_img.convert("RGB")
    buf_clean = io.BytesIO()
    clean_final.save(buf_clean, format="JPEG", quality=quality, optimize=True)
    data_clean = base64.b64encode(buf_clean.getvalue()).decode()

    grid1 = _render_grid_variant(base_img, ow, oh, sw, sh, mx, my, display,
                                  grid_step=grid_step, line_color=(0, 255, 180, 235),
                                  label_every=100)
    buf1 = io.BytesIO(); grid1.save(buf1, format="JPEG", quality=quality, optimize=True)
    data_grid1 = base64.b64encode(buf1.getvalue()).decode()

    grid2 = _render_grid_variant(base_img, ow, oh, sw, sh, mx, my, display,
                                  grid_step=20, line_color=(255, 60, 60, 235),
                                  label_every=60, major_only_labels=True)
    buf2 = io.BytesIO(); grid2.save(buf2, format="JPEG", quality=max(quality, 78), optimize=True)
    data_grid2 = base64.b64encode(buf2.getvalue()).decode()

    return {
        "data": data_clean,
        "data_grid": data_grid1,
        "data_grid2": data_grid2,
        "data_grid3": data_grid2,  # توافقية رجعية: نفس قيمة data_grid2 لأي كود قديم يقرأ data_grid3
        "width": ow, "height": oh, "mouse_x": mx, "mouse_y": my,
    }


def _frame_hash(data: str) -> str:
    """
    إصلاح: كان يُحسب على أول 2000 حرف فقط من بيانات الصورة، مما يعني أي تغيير
    بصري يقع خارج الجزء الممثَّل ضمن تلك الأحرف الأولى (مثل تحديد نص شريط العنوان
    بعد Ctrl+L، أو أي تغيير طفيف/في منطقة لا تتوافق مع بداية ترميز base64) لا
    يُكتشف أبداً، فتُحجب الصورة الجديدة عبر delta suppression رغم اختلافها فعلياً
    عن السابقة، ولا تصل أي لقطة محدَّثة للعميل. الحل: حساب MD5 على كامل البيانات
    بدل تقطيعها — هذا سريع جداً (أقل من مللي ثانية حتى لصور كبيرة) فلا تكلفة أداء
    حقيقية، ويضمن اكتشاف أي تغيير بصري حقيقي في أي مكان من الصورة.
    """
    return hashlib.md5(data.encode()).hexdigest()


# ════════════════════════════════════════════════════════════════
# ── SafeSearch ───────────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

_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 Execution (per-display) ─────────────────────────
# ════════════════════════════════════════════════════════════════

_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, display: str, timeout: int = 60) -> dict:
    """ينفّذ أمر bash مع DISPLAY خاص بالمستخدم."""
    cmd = _sanitize_pkill(_safe_search(cmd))
    async with _terminal_sem:
        env = {**os.environ, "DISPLAY": display,
               "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)


# ════════════════════════════════════════════════════════════════
# ── xdotool helpers (per-display) ────────────────────────────
# ════════════════════════════════════════════════════════════════

async def xdo(args: list, display: str, timeout: int = 10) -> dict:
    env = {**os.environ, "DISPLAY": display}
    def _run():
        r = subprocess.run(
            ["xdotool"] + args, env=env,
            timeout=timeout, capture_output=True, text=True
        )
        return {"rc": r.returncode, "out": r.stdout, "err": r.stderr}
    return await asyncio.to_thread(_run)


async def type_smart(text: str, display: str) -> dict:
    """كتابة نص ذكية بـ display محدد."""
    has_arabic = bool(re.search(r'[\u0600-\u06FF]', text))
    env = {**os.environ, "DISPLAY": display}

    if has_arabic:
        def _paste():
            p = subprocess.Popen(
                ["xclip", "-selection", "clipboard"],
                stdin=subprocess.PIPE, env=env
            )
            p.communicate(text.encode("utf-8"))
        await asyncio.to_thread(_paste)
        await asyncio.sleep(0.15)
        await xdo(["key", "--clearmodifiers", "ctrl+v"], display)
        return {"method": "clipboard+paste"}
    else:
        r = await xdo(["type", "--clearmodifiers", "--delay", "25", text], display)
        return {"method": "xdotool", "rc": r["rc"]}


# ════════════════════════════════════════════════════════════════
# ── Search Sources ────────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

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": "Google News RSS",
         "cmd": f"curl -sL --max-time 15 'https://news.google.com/rss/search?q={q}&hl=ar&gl=AR&ceid=AR:ar' | python3 -c \"import sys,re;x=sys.stdin.read();t=re.findall(r'<title><!\\[CDATA\\[(.*?)\\]\\]></title>|<title>(.*?)</title>',x);[(print(str(i+1)+'. '+(a or b).strip()[:160])) for i,(a,b) in enumerate(t[1:8]) if (a or b).strip()]\""},
        {"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()


async def run_cmd_smart(cmd: str, display: str, timeout: int = 60) -> dict:
    res = await run_cmd(cmd, display, 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

    sources = _search_sources(query)
    results = []
    for src in sources:
        r2 = await run_cmd(src["cmd"], display, 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 res


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

app = FastAPI(title="Z-Computer-Mode v11 — Per-User Isolation")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True,
                   allow_methods=["*"], allow_headers=["*"])


@app.get("/", response_class=HTMLResponse)
async def root():
    n = len(_user_displays)
    users_info = ""
    with _display_lock:
        for uid, sess in _user_displays.items():
            ws_status = "🟢 متصل" if sess.get("active_ws") else "⚫ غير متصل"
            users_info += f"<tr><td>{uid[:30]}...</td><td>{sess['display']}</td><td>{ws_status}</td></tr>"

    return f"""<!DOCTYPE html><html><head><meta charset="utf-8">
<title>Z Computer Mode v11</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 v11</h1>
<p class="ok">✅ Server RUNNING — Per-User Isolated Displays</p>
<p>🌐 Browser: <b>{BROWSER}</b></p>
<p>👥 Active users: <b>{n}</b></p>
<p>📋 Display range: <b>:{DISPLAY_BASE} → :{DISPLAY_MAX}</b></p>
<hr style="border-color:#333">
<h3>👤 User Sessions</h3>
<table><tr><th>User ID</th><th>Display</th><th>Status</th></tr>
{users_info or '<tr><td colspan="3" style="text-align:center;color:#666">No active sessions</td></tr>'}
</table>
<hr style="border-color:#333">
<p class="info">Endpoints: /health · /ws?user_id=email (WebSocket)</p>
</body></html>"""


# ════════════════════════════════════════════════════════════════
# ── Action Handler (per-session) ─────────────────────────────
# ════════════════════════════════════════════════════════════════

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

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

    # ── إصلاح جذري: قفل خاص بكل session لمنع تراكم/تضارب عمليات capture المتزامنة ──
    # المشكلة الأصلية: action == "screenshot" كان يُنفَّذ بـ await مباشر ضمن حلقة
    # while True الرئيسية في websocket_endpoint، فإذا تأخرت capture_with_grid (قد تصل
    # لدقيقة ونصف في أسوأ سيناريو فشل/تأخر scrot+import+ffmpeg)، تبقى الحلقة كلها
    # محجوبة ولا تستقبل أي رسالة عميل جديدة (نقرة، طلب screenshot آخر، إلخ) حتى تنتهي.
    # الحل: كل طلب screenshot يُشغَّل فوراً عبر create_task (لا يحجب الحلقة أبداً).
    #
    # ── إصلاح ثانٍ (مهم جداً): طلب screenshot الصريح من العميل له أولوية مطلقة ──
    # المشكلة المكتشفة لاحقاً: لو استُخدم قفل واحد مشترك بين shot_bg (تلقائية، بعد
    # كل فعل كنقرة/فتح تبويب) و shot_explicit (صريحة، يطلبها العميل مباشرة بعد كل
    # خطوة لتحديث الصورة المعروضة)، فإن shot_explicit قد تنتظر خلف shot_bg طويلاً
    # إذا كانت الأخيرة قد بدأت فعلاً وعلقت داخل capture_with_grid (محاولات فاشلة
    # متتالية). هذا يجعل العميل يرى "لم تصل لقطة شاشة" بشكل متكرر حتى بعد نجاح
    # الخطوة الفعلية (مثل open_tab)، لأن طلبه العاجل كان يصطف خلف عملية تلقائية بطيئة.
    # الحل: shot_explicit (الصريحة فقط) تحاول الحصول على القفل لفترة قصيرة جداً
    # (1 ثانية)، وإن لم تنجح (لأن shot_bg تستخدمه)، تُنفَّذ التصوير مباشرة بدون قفل
    # بدل الانتظار — لأن استجابة العميل الفورية أهم من تفادي تزاحم CPU عرضي بسيط،
    # وقراءة الشاشة (X11) عملية قراءة فقط لا تُسبب أي تلف بيانات عند التزاحم.
    if "_shot_lock" not in sess:
        sess["_shot_lock"] = asyncio.Lock()
    _shot_lock = sess["_shot_lock"]

    async def _safe_capture(scale, quality, force_mx=None, force_my=None):
        """
        تُستخدم من shot_bg (التلقائية) — تنتظر القفل ثم تنتظر التقاط الشاشة
        حتى ينتهي فعلياً، بدون أي سقف زمني يقطعها. هذا يضمن أن العملية لن
        تُقطع أبداً في منتصفها وتُرجع فراغاً؛ ستكمل حتى تنجح (أو تفشل كل
        المحاولات الداخلية في capture_with_grid وتُرجع placeholder صريح).
        """
        async with _shot_lock:
            return await asyncio.to_thread(
                capture_with_grid, display, scale, quality, force_mx, force_my
            )

    async def _priority_capture(scale, quality, force_mx=None, force_my=None):
        """
        تُستخدم من shot_explicit (الطلب الصريح من العميل) فقط — أولوية قصوى.
        تحاول الحصول على القفل لمدة قصيرة (1 ثانية) فقط؛ إن لم تنجح (القفل
        محجوز من shot_bg تلقائية بطيئة)، تُنفَّذ التصوير فوراً بدون قفل بدل
        الانتظار خلف عملية أخرى — العميل يجب أن يحصل على رد سريع دائماً.
        بعد الحصول على القفل (أو تجاوزه)، لا يوجد أي سقف زمني على عملية
        الالتقاط نفسها — تنتظر حتى تكتمل فعلياً بدل أن تُقطع في المنتصف.
        """
        try:
            await asyncio.wait_for(_shot_lock.acquire(), timeout=1.0)
            try:
                return await asyncio.to_thread(
                    capture_with_grid, display, scale, quality, force_mx, force_my
                )
            finally:
                _shot_lock.release()
        except asyncio.TimeoutError:
            # لم نحصل على القفل بسرعة كافية — ننفّذ التصوير مباشرة بدون قفل
            # بدل الانتظار خلف عملية أخرى، لكن بدون أي سقف زمني على الالتقاط
            return await asyncio.to_thread(
                capture_with_grid, display, scale, quality, force_mx, force_my
            )

    async def shot_bg(label: str = "", delay: float = 0.5,
                      force_mx: int | None = None, force_my: int | None = None,
                      extra_shot_delay: float | None = None):
        """
        auto_shot في الخلفية — مع rate limit + delta suppression.
        إذا تم تمرير extra_shot_delay، تُرسل لقطة ثانية إضافية بعد ذلك التأخير
        الإضافي (محسوباً من وقت انتهاء اللقطة الأولى) — هذا يغطي حالة النقر على
        رابط ينقل لصفحة جديدة قد لا تكتمل تحميلها خلال التأخير الأولي القصير
        (مثلاً صفحة بطيئة على سيرفر محدود الموارد)، دون الحاجة لمعرفة مسبقة بنوع
        العنصر المنقور عليه. تتجاهل delta suppression لهذه اللقطة الثانية تحديداً
        لأن المحتوى متوقع أن يكون مختلفاً (صفحة تحمّلت أكثر) حتى لو تشابه الـ hash
        جزئياً مع خلفية مشابهة.
        """
        try:
            await asyncio.sleep(delay)
            now = time.time()
            if now - sess.get("last_bg_shot_ts", 0) < 0.3:
                await asyncio.sleep(0.3 - (now - sess["last_bg_shot_ts"]))

            result = await _safe_capture(0.65, 72, force_mx, force_my)
            sess["last_bg_shot_ts"] = time.time()
            if result["data"]:
                fh = _frame_hash(result["data"])
                if fh != sess.get("last_bg_hash", ""):
                    sess["last_bg_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": True, "label": label,
                        "screen_width": result["width"],
                        "screen_height": result["height"],
                        "mouse_x": result["mouse_x"],
                        "mouse_y": result["mouse_y"],
                        "has_grid": True,
                    })

            if extra_shot_delay:
                await asyncio.sleep(extra_shot_delay)
                result2 = await _safe_capture(0.65, 72, force_mx, force_my)
                sess["last_bg_shot_ts"] = time.time()
                if result2["data"]:
                    fh2 = _frame_hash(result2["data"])
                    sess["last_bg_hash"] = fh2
                    await send({
                        "type": "screenshot",
                        "data": result2["data"],
                        "data_grid": result2.get("data_grid", ""),
                        "data_grid2": result2.get("data_grid2", ""),
                        "data_grid3": result2.get("data_grid3", ""),
                        "ts": int(time.time() * 1000),
                        "auto": True, "label": f"{label} (delayed)",
                        "screen_width": result2["width"],
                        "screen_height": result2["height"],
                        "mouse_x": result2["mouse_x"],
                        "mouse_y": result2["mouse_y"],
                        "has_grid": True,
                    })
        except Exception as e:
            # لقطة خلفية فاشلة لا يجب أن توقف الجلسة أو تظهر بصمت في اللوق فقط.
            print(f"[shot_bg:{display}] ⚠️ {e}")

    async def shot_explicit(label: str = ""):
        """screenshot صريح — يُرسل دائماً بدون delta suppression، وله أولوية مطلقة
        على أي عملية shot_bg تلقائية جارية (لا ينتظر خلفها طويلاً).
        ── إصلاح جذري: try/except شامل + إرسال رسالة خطأ صريحة للعميل ──
        المشكلة القديمة: هذه الدالة تُشغَّل عبر asyncio.create_task بدون أي
        معالجة استثناء محيطة بها. أي خطأ غير متوقع كان يُسقط الـ task بالكامل
        بصمت تام — لا رسالة تصل للعميل ولا حتى سطر في اللوق، فيبقى العميل
        ينتظر frame لن يصل أبداً. الآن أي فشل يُسجَّل ويُرسَل كرسالة "error"
        صريحة، والعميل (index.html) يتعامل معها ويعيد المحاولة فوراً."""
        try:
            result = await _priority_capture(0.65, 75)
            if not result or not result["data"]:
                print(f"[shot_explicit:{display}] ⚠️ empty capture result")
                await send({"type": "error", "action": "screenshot", "msg": "capture returned empty"})
                return
            sess["last_bg_hash"] = _frame_hash(result["data"])
            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": False, "label": label,
                "screen_width": result["width"],
                "screen_height": result["height"],
                "mouse_x": result["mouse_x"],
                "mouse_y": result["mouse_y"],
                "has_grid": True,
            })
        except Exception as e:
            print(f"[shot_explicit:{display}] ❌ EXCEPTION: {e}")
            try:
                await send({"type": "error", "action": "screenshot", "msg": str(e)})
            except Exception:
                pass

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

    # ── screenshot ────────────────────────────────────────────
    # إصلاح: create_task بدل await مباشر — لا يحجب حلقة استقبال الرسائل أبداً،
    # حتى لو تأخرت capture_with_grid لأي سبب (CPU ضعيف، X11 بطيء، إلخ).
    # هذا يضمن أن أي إجراء لاحق (نقرة، طباعة، طلب screenshot آخر) يصل ويُعالَج
    # فوراً دون انتظار اكتمال هذه اللقطة أولاً.
    if action == "screenshot":
        # ── طبقة حماية إضافية: حتى لو حصل خطأ غير متوقع تماماً داخل
        # shot_explicit تفلت من الـ try/except الداخلي (مثل CancelledError
        # أو خطأ عند إنشاء الـ task نفسه)، هذا الغلاف الخارجي يضمن تسجيله
        # في اللوق بدل أن يختفي بصمت كـ "unhandled task exception" في event
        # loop بايثون (وهذا كان له سلوك افتراضي بصامت في بعض إعدادات uvicorn).
        _task = asyncio.create_task(shot_explicit("explicit screenshot"))
        def _on_shot_done(t: asyncio.Task):
            exc = t.exception() if not t.cancelled() else None
            if exc:
                print(f"[screenshot-task:{display}] ❌ unhandled exception: {exc}")
        _task.add_done_callback(_on_shot_done)

    # ── terminal ─────────────────────────────────────────────
    elif 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, display, int(data.get("timeout", 60)))
        await send({
            "type": "terminal_result",
            "cmd": cmd,
            "stdout": res["stdout"],
            "stderr": res.get("stderr", ""),
            "returncode": res["returncode"],
        })
        # ── تعطيل الالتقاط التلقائي: لا تُؤخذ لقطة شاشة إلا بطلب صريح ──

    # ── mouse_move ───────────────────────────────────────────
    elif action == "mouse_move":
        x, y = int(data.get("x", 0)), int(data.get("y", 0))
        await xdo(["mousemove", "--sync", str(x), str(y)], display)
        await send({"type": "ack", "action": "mouse_move", "x": x, "y": y})
        # ── تعطيل الالتقاط التلقائي ──

    # ── mouse_click ──────────────────────────────────────────
    elif action == "mouse_click":
        x, y   = int(data.get("x", 0)), int(data.get("y", 0))
        btn    = {"left": "1", "middle": "2", "right": "3"}.get(data.get("button", "left"), "1")
        double = data.get("double", False)
        await xdo(["mousemove", "--sync", str(x), str(y)], display)
        await asyncio.sleep(0.07)
        if double:
            await xdo(["click", "--repeat", "2", "--delay", "100", btn], display)
        else:
            await xdo(["click", btn], display)
        btn_name = {"1": "left", "2": "middle", "3": "right"}.get(btn, "left")
        await send({"type": "ack", "action": "mouse_click", "x": x, "y": y, "button": btn_name})
        # ── تعطيل الالتقاط التلقائي ──

    # ── mouse_drag ───────────────────────────────────────────
    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))
        await xdo(["mousemove", str(x1), str(y1)], display)
        await xdo(["mousedown", "1"], display)
        await asyncio.sleep(0.1)
        await xdo(["mousemove", str(x2), str(y2)], display)
        await asyncio.sleep(0.1)
        await xdo(["mouseup", "1"], display)
        await send({"type": "ack", "action": "mouse_drag"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── keyboard_type ────────────────────────────────────────
    elif action == "keyboard_type":
        text = _safe_search(data.get("text", ""))
        if text:
            res = await type_smart(text, display)
            await send({"type": "ack", "action": "keyboard_type", "method": res["method"]})
        # ── تعطيل الالتقاط التلقائي ──

    # ── keyboard_hotkey ──────────────────────────────────────
    elif action == "keyboard_hotkey":
        keys = data.get("keys", [])
        if keys:
            await xdo(["key", "--clearmodifiers", "+".join(keys)], display)
        await send({"type": "ack", "action": "keyboard_hotkey", "keys": keys})
        # ── تعطيل الالتقاط التلقائي ──

    # ── keyboard_press ───────────────────────────────────────
    elif action == "keyboard_press":
        key = data.get("key", "")
        if key:
            await xdo(["key", "--clearmodifiers", key], display)
        await send({"type": "ack", "action": "keyboard_press"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── scroll ───────────────────────────────────────────────
    elif action == "scroll":
        x, y   = int(data.get("x", 960)), int(data.get("y", 540))
        clicks = max(-5, min(5, int(data.get("clicks", 3))))
        btn    = "4" if clicks > 0 else "5"
        await xdo(["mousemove", str(x), str(y)], display)
        for _ in range(abs(clicks)):
            await xdo(["click", btn], display)
            await asyncio.sleep(0.025)
        await send({"type": "ack", "action": "scroll", "clicks": clicks})
        # ── تعطيل الالتقاط التلقائي ──

    # ── clipboard_write ──────────────────────────────────────
    elif action == "clipboard_write":
        text = data.get("text", "")
        env  = {**os.environ, "DISPLAY": display}
        def _clip():
            p = subprocess.Popen(["xclip", "-selection", "clipboard"], stdin=subprocess.PIPE, env=env)
            p.communicate(text.encode("utf-8"))
        await asyncio.to_thread(_clip)
        await send({"type": "ack", "action": "clipboard_write", "length": len(text)})

    # ── clipboard_read ───────────────────────────────────────
    elif action == "clipboard_read":
        res = await run_cmd("xclip -selection clipboard -o", display, 5)
        await send({"type": "clipboard_content", "text": res["stdout"]})

    # ── paste ────────────────────────────────────────────────
    elif action == "paste":
        text = data.get("text", "")
        if text:
            env  = {**os.environ, "DISPLAY": display}
            def _clip2():
                p = subprocess.Popen(["xclip", "-selection", "clipboard"], stdin=subprocess.PIPE, env=env)
                p.communicate(text.encode("utf-8"))
            await asyncio.to_thread(_clip2)
            await asyncio.sleep(0.1)
            await xdo(["key", "--clearmodifiers", "ctrl+v"], display)
        await send({"type": "ack", "action": "paste"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── open_app ─────────────────────────────────────────────
    elif action == "open_app":
        cmd = _safe_search(data.get("cmd", ""))
        if not cmd:
            await send({"type": "ack", "action": "open_app"})
            return
        if "firefox" in cmd.lower():
            profile_dir = _ensure_firefox_profile(sess["user_id"], display)
            # قتل فوري وقوي (SIGKILL) بدل pkill اللطيف: يمنع فايرفوكس من
            # "التقاط" حالة نصف-مفتوحة تُسجَّل لاحقاً كتحطّم (crash) يُنتج
            # شاشة "Restore Session" في المرة القادمة.
            await run_cmd(
                "pkill -9 -f '[f]irefox' 2>/dev/null; sleep 0.6; echo CLEANED",
                display, timeout=8
            )
            _wipe_firefox_session_data(profile_dir)
            # حقن --profile داخل أمر فايرفوكس (إن لم يكن محقوناً مسبقاً)
            if "--profile" not in cmd and "-P " not in cmd:
                flags_str = " ".join(FIREFOX_CLI_FLAGS)
                cmd = cmd.replace("firefox", f"firefox --profile '{profile_dir}' {flags_str}", 1)
        env = {**os.environ, "DISPLAY": display, **(FIREFOX_ENV_EXTRA if "firefox" in cmd.lower() else {})}
        proc = subprocess.Popen(cmd, shell=True, env=env,
                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        if any(b in cmd for b in ("firefox", "chromium", "chrome")):
            with _display_lock:
                if sess["user_id"] in _user_displays:
                    _user_displays[sess["user_id"]]["browser_proc"] = proc
        await send({"type": "ack", "action": "open_app", "cmd": cmd})
        # ── تعطيل الالتقاط التلقائي بعد فتح التطبيق ──

    # ── open_browser ─────────────────────────────────────────
    elif action == "open_browser":
        url = _safe_search(data.get("url", "") or "about:blank")
        profile_dir = _ensure_firefox_profile(sess["user_id"], display)
        # قتل فوري وقوي (SIGKILL) بدل pkill اللطيف — يمنع تسجيل حالة
        # "تحطّم" في sessionstore، وهي السبب الجذري لشاشة "Restore Session"
        # التي تحجب المحتوى الفعلي وتُلتقط كلقطة شاشة صالحة (غير سوداء)
        # لكنها ليست الصفحة المطلوبة.
        await run_cmd(
            "pkill -9 -f '[f]irefox' 2>/dev/null; sleep 0.6; echo CLEANED",
            display, timeout=8
        )
        _wipe_firefox_session_data(profile_dir)
        env = {**os.environ, "DISPLAY": display, **FIREFOX_ENV_EXTRA}
        proc = subprocess.Popen(
            [BROWSER, "--profile", profile_dir, *FIREFOX_CLI_FLAGS, url], env=env,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
        )
        with _display_lock:
            if sess["user_id"] in _user_displays:
                _user_displays[sess["user_id"]]["browser_proc"] = proc
        await send({"type": "ack", "action": "open_browser", "url": url})
        # ── تعطيل الالتقاط التلقائي بعد فتح المتصفح ──

    # ── open_tab ─────────────────────────────────────────────
    elif action == "open_tab":
        # إصلاح: زيادة التأخيرات بين كل خطوة فرعية هنا (Ctrl+T → Ctrl+L → كتابة الرابط)
        # المشكلة الأصلية: التأخير القصير (0.4s) بعد Ctrl+T لم يكن كافياً دائماً، خصوصاً
        # على سيرفر محدود الموارد قد يستغرق فايرفوكس فيه وقتاً أطول لإنشاء التبويب الجديد
        # فعلياً وتحويل التركيز (focus) إليه. إذا وصل Ctrl+L والكتابة قبل اكتمال ذلك،
        # فإنها قد تذهب للتبويب القديم بالخطأ، فلا تنتقل الصفحة فعلياً رغم نجاح الأمر
        # ظاهرياً (بدون أي رسالة خطأ)، فيبقى المستخدم يرى نفس الصفحة القديمة للأبد.
        url = _safe_search(data.get("url", "") or "about:blank")
        await xdo(["key", "--clearmodifiers", "ctrl+t"], display)
        await asyncio.sleep(0.8)
        await xdo(["key", "--clearmodifiers", "ctrl+l"], display)
        await asyncio.sleep(0.3)
        await type_smart(url, display)
        await asyncio.sleep(0.3)
        await xdo(["key", "--clearmodifiers", "Return"], display)
        await send({"type": "ack", "action": "open_tab", "url": url})
        # ── تعطيل الالتقاط التلقائي ──

    # ── close_tab ────────────────────────────────────────────
    elif action == "close_tab":
        await xdo(["key", "--clearmodifiers", "ctrl+w"], display)
        await send({"type": "ack", "action": "close_tab"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── browser_back ─────────────────────────────────────────
    elif action == "browser_back":
        await xdo(["key", "--clearmodifiers", "alt+Left"], display)
        await send({"type": "ack", "action": "browser_back"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── browser_forward ──────────────────────────────────────
    elif action == "browser_forward":
        await xdo(["key", "--clearmodifiers", "alt+Right"], display)
        await send({"type": "ack", "action": "browser_forward"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── browser_search ───────────────────────────────────────
    elif action == "browser_search":
        url = _safe_search(data.get("url", "") or data.get("query", ""))
        await xdo(["key", "--clearmodifiers", "ctrl+l"], display)
        await asyncio.sleep(0.2)
        await type_smart(url, display)
        await asyncio.sleep(0.15)
        await xdo(["key", "--clearmodifiers", "Return"], display)
        await send({"type": "ack", "action": "browser_search"})
        # ── تعطيل الالتقاط التلقائي ──

    # ── screen_info ──────────────────────────────────────────
    elif action == "screen_info":
        w, h   = await asyncio.to_thread(_get_screen_size, display)
        mx, my = await asyncio.to_thread(_get_mouse_pos, display)
        await send({
            "type": "screen_info",
            "width": w, "height": h,
            "mouse_x": mx, "mouse_y": my,
            "browser": BROWSER,
            "display": display,
        })

    # ── unknown ──────────────────────────────────────────────
    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()

    # الحصول على session المستخدم أو إنشاء واحدة جديدة
    sess = await get_or_create_user_session(user_id, ws)

    # تأكد أن Xvfb يعمل لهذا المستخدم
    await ensure_xvfb_for_session(sess)

    display = sess["display"]

    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:
        w, h = await asyncio.to_thread(_get_screen_size, display)
        await ws.send_text(json.dumps({
            "type": "connected",
            "screen_width": w, "screen_height": h,
            "browser": BROWSER,
            "display": display,
            "user_id": user_id,
            "session_id": id(ws),
            "msg": f"Z Computer Mode v11 | User: {user_id} | Display: {display} | Browser: {BROWSER} | Screen: {w}x{h}",
        }, ensure_ascii=False))

        # لقطة شاشة أولية
        result = await asyncio.to_thread(capture_with_grid, display, 0.65, 72)
        if result["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_bg_hash"] = _frame_hash(result["data"])

    except Exception as e:
        print(f"[ws:{user_id}] init error: {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}] error: {e}")
    finally:
        hb_task.cancel()
        await destroy_user_ws(user_id, ws)


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

@app.get("/screenshot")
async def rest_screenshot(user_id: str = "anonymous"):
    with _display_lock:
        sess = _user_displays.get(user_id)
    display = sess["display"] if sess else f":{DISPLAY_BASE}"
    result = await asyncio.to_thread(capture_with_grid, display, 0.7, 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,
        "display": display,
        "user_id": user_id,
    })

@app.post("/terminal")
async def rest_terminal(body: dict):
    user_id = body.get("user_id", "anonymous")
    with _display_lock:
        sess = _user_displays.get(user_id)
    display = sess["display"] if sess else f":{DISPLAY_BASE}"
    return JSONResponse(await run_cmd_smart(body.get("cmd", ""), display, body.get("timeout", 60)))

@app.get("/health")
async def health():
    with _display_lock:
        n = len(_user_displays)
        users = [
            {"user_id": uid, "display": s["display"],
             "connected": bool(s.get("active_ws"))}
            for uid, s in _user_displays.items()
        ]
    return {
        "status": "ok",
        "version": "v13-fast-capture-fixed-firefox",
        "browser": BROWSER,
        "active_users": n,
        "users": users,
    }


# ════════════════════════════════════════════════════════════════
# ── Background Tasks ──────────────────────────────────────────
# ════════════════════════════════════════════════════════════════

async def _cleanup_tmp():
    """ينظّف /tmp كل 5 دقائق. ملاحظة v13: محرك الالتقاط
    الجديد (import-pipe/xlib-direct) لا يكتب أي ملفات zss_* على الإطلاق (كلشيء
    في الذاكرة عبر pipe)، فهذه الدالة أصبحت غير ضرورية فعلياً للقطات
    الشاشة، لكنها تُرك كشبكة أمان إضافية (مثلاً لو أضيف كود مستقبلاً
    يكتب ملفات مؤقتة بنفس البادئة)."""
    while True:
        await asyncio.sleep(300)
        try:
            subprocess.run(
                ["find", "/tmp", "-name", "zss_*", "-mmin", "+10", "-delete"],
                capture_output=True, timeout=10
            )
        except Exception as e:
            print(f"[cleanup] {e}")


async def _cleanup_idle_sessions():
    """
    يُزيل sessions المستخدمين غير النشطين (لا اتصال منذ أكثر من ساعة)
    لتحرير الـ displays والذاكرة.
    """
    while True:
        await asyncio.sleep(1800)  # كل 30 دقيقة
        now = time.time()
        to_remove = []
        with _display_lock:
            for uid, sess in list(_user_displays.items()):
                if sess.get("active_ws"):
                    continue  # لا تحذف sessions النشطة
                if now - sess.get("created", now) > 3600:  # أكثر من ساعة
                    to_remove.append((uid, sess))

        for uid, sess in to_remove:
            print(f"[cleanup] 🗑️ Removing idle session for '{uid}' on {sess['display']}")
            _kill_proc(sess.get("browser_proc"))
            _kill_proc(sess.get("xvfb_proc"))
            _kill_display_processes(sess["display"])
            with _display_lock:
                _user_displays.pop(uid, None)
                try:
                    _display_numbers.discard(int(sess["display"].lstrip(":")))
                except Exception:
                    pass


@app.on_event("startup")
async def startup():
    asyncio.create_task(_cleanup_tmp())
    asyncio.create_task(_cleanup_idle_sessions())
    print("✅ Z Computer Mode v11 ready — Per-User Isolated Displays")
    print(f"   Display range: :{DISPLAY_BASE} → :{DISPLAY_MAX}")
    print(f"   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")