Blame for server.py

"""
Telearn Arcade Hub — FastAPI server.

Serves the project catalog, static assets, and WebSocket bridge that streams
headless Pygame/OpenCV student projects to the browser.
"""
from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
import logging
import os
import signal
import socket
import shutil
import stat
import subprocess
import sys
import tempfile
import time
import secrets
import re
from collections import deque
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urlsplit

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from classroom_service import router as classroom_router
from admin_service import router as admin_router

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECTS_DIR = os.path.join(BASE_DIR, "projects")
TEMPLATES_DIR = os.path.join(BASE_DIR, "templates")
STATIC_DIR = os.path.join(BASE_DIR, "static")
MODELS_DIR = os.path.join(BASE_DIR, "models")

MAX_CONCURRENT_RUNNERS = int(os.environ.get("MAX_CONCURRENT_RUNNERS", "3"))
SESSION_TIMEOUT_SEC = int(os.environ.get("SESSION_TIMEOUT_SEC", "900"))  # 15 min
IDLE_TIMEOUT_SEC = max(1, int(os.environ.get("IDLE_TIMEOUT_SEC", "720")))  # 12 min
IDLE_WARNING_SEC = min(
    IDLE_TIMEOUT_SEC, max(0, int(os.environ.get("IDLE_WARNING_SEC", "600")))
)  # warn after 10 min idle
HEARTBEAT_TIMEOUT_SEC = max(
    30, int(os.environ.get("HEARTBEAT_TIMEOUT_SEC", "75"))
)
PROJECT_CACHE_TTL_SEC = float(os.environ.get("PROJECT_CACHE_TTL_SEC", "10"))
RUNNER_CONNECT_TIMEOUT_SEC = float(os.environ.get("RUNNER_CONNECT_TIMEOUT_SEC", "8"))
RUNNER_TMP_DIR = os.environ.get("RUNNER_TMP_DIR", "/tmp/telearn_runner")
RUNNER_MEMORY_LIMIT_MB = max(512, int(os.environ.get("RUNNER_MEMORY_LIMIT_MB", "3072")))
RUNNER_CPU_LIMIT_SEC = max(30, int(os.environ.get("RUNNER_CPU_LIMIT_SEC", "900")))
RUNNER_TMP_LIMIT_MB = max(64, int(os.environ.get("RUNNER_TMP_LIMIT_MB", "256")))
RUNNER_PROJECT_LIMIT_MB = max(8, int(os.environ.get("RUNNER_PROJECT_LIMIT_MB", "128")))
RUNNER_PROJECT_FILE_LIMIT = max(20, int(os.environ.get("RUNNER_PROJECT_FILE_LIMIT", "2000")))
RUNNER_STARTS_PER_IP_PER_HOUR = max(1, int(os.environ.get("RUNNER_STARTS_PER_IP_PER_HOUR", "120")))
RUNNER_START_COOLDOWN_SEC = max(0.5, float(os.environ.get("RUNNER_START_COOLDOWN_SEC", "0.75")))
ENABLE_HSTS = os.environ.get("ENABLE_HSTS", "false").lower() in {"1", "true", "yes"}
HSTS_MAX_AGE_SEC = max(0, int(os.environ.get("HSTS_MAX_AGE_SEC", "31536000")))
BWRAP_PATH = shutil.which("bwrap")
PRLIMIT_PATH = shutil.which("prlimit")
RUNNER_ISOLATION_MODE = os.environ.get("RUNNER_ISOLATION_MODE", "sandbox").lower()

os.makedirs(PROJECTS_DIR, exist_ok=True)
os.makedirs(TEMPLATES_DIR, exist_ok=True)
os.makedirs(STATIC_DIR, exist_ok=True)
os.makedirs(MODELS_DIR, exist_ok=True)
os.makedirs(RUNNER_TMP_DIR, mode=0o700, exist_ok=True)
os.chmod(RUNNER_TMP_DIR, 0o700)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [server] %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger("telearn.server")

app = FastAPI(title="Student Pygame/OpenCV Arcade Hub")
templates = Jinja2Templates(directory=TEMPLATES_DIR)
app.mount("/classroom-assets", StaticFiles(directory=os.path.join(BASE_DIR, "classroom_static")), name="classroom-assets")
app.include_router(classroom_router)
app.include_router(admin_router)


@app.middleware("http")
async def browser_runtime_headers(request: Request, call_next):
    """Set browser isolation and response security policies."""
    nonce = secrets.token_urlsafe(18)
    request.scope.setdefault("state", {})["csp_nonce"] = nonce
    response = await call_next(request)
    response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
    response.headers["Cross-Origin-Embedder-Policy"] = "credentialless"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    response.headers["Permissions-Policy"] = "camera=(self), microphone=(), geolocation=(), payment=(), usb=()"
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; base-uri 'self'; object-src 'none'; "
        f"script-src 'self' 'nonce-{nonce}' 'wasm-unsafe-eval'; "
        "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
        "img-src 'self' data: blob: https://api.qrserver.com; "
        "font-src 'self' data: https://fonts.gstatic.com; "
        "connect-src 'self' ws: wss:; "
        "worker-src 'self' blob:; "
        "frame-src 'self' https://appetize.io https://*.appetize.io https://makecode.com https://*.makecode.com; "
        "frame-ancestors 'self' https://huggingface.co; form-action 'self'"
    )
    # Student-authored web projects need JavaScript, but must remain an
    # opaque-origin sandbox even when opened directly instead of in the player.
    if request.url.path.startswith("/static_projects/"):
        response.headers["Content-Security-Policy"] = (
            "sandbox allow-scripts allow-forms allow-pointer-lock; "
            "default-src 'self' data: blob: https:; "
            "script-src 'self' 'unsafe-inline' 'unsafe-eval' https:; "
            "style-src 'self' 'unsafe-inline' https:; "
            "img-src 'self' data: blob: https:; font-src 'self' data: https:; "
            "connect-src 'self' https: wss:; media-src 'self' data: blob: https:; "
            "object-src 'none'; base-uri 'self'; form-action 'self'; frame-ancestors 'self'"
        )
    if HSTS_MAX_AGE_SEC and (ENABLE_HSTS or request.url.scheme == "https"):
        response.headers["Strict-Transport-Security"] = f"max-age={HSTS_MAX_AGE_SEC}"
    return response

# Concurrency + live session metrics
runner_semaphore = asyncio.Semaphore(MAX_CONCURRENT_RUNNERS)
active_sessions: dict[str, dict[str, Any]] = {}
_sessions_lock = asyncio.Lock()
_metrics = {
    "sessions_started": 0,
    "sessions_rejected": 0,
    "sessions_timed_out": 0,
    "sessions_idle_timed_out": 0,
    "sessions_heartbeat_timed_out": 0,
    "sessions_failed_connect": 0,
}

# Project tree cache
_project_cache: dict[str, Any] = {"tree": None, "built_at": 0.0, "signature": None}
_runner_start_windows: dict[str, deque] = {}


def allow_runner_start(remote_ip: str) -> bool:
    """Limit public server-side Python starts per client IP across sessions."""
    current = time.monotonic()
    window = _runner_start_windows.get(remote_ip)
    if window is None:
        window = deque()
        _runner_start_windows[remote_ip] = window
    while window and current - window[0] >= 3600:
        window.popleft()
    if window and current - window[-1] < RUNNER_START_COOLDOWN_SEC:
        return False
    if len(window) >= RUNNER_STARTS_PER_IP_PER_HOUR:
        return False
    window.append(current)
    if len(_runner_start_windows) > 4096:
        for key, old_window in list(_runner_start_windows.items()):
            while old_window and current - old_window[0] >= 3600:
                old_window.popleft()
            if not old_window:
                _runner_start_windows.pop(key, None)
            if len(_runner_start_windows) <= 3072:
                break
    return True


def _session_idle_state(
    idle_for_sec: float,
    warning_sec: float = IDLE_WARNING_SEC,
    timeout_sec: float = IDLE_TIMEOUT_SEC,
) -> str:
    """Return active, warning, or timeout for a session's idle duration."""
    if idle_for_sec >= timeout_sec:
        return "timeout"
    if warning_sec < timeout_sec and idle_for_sec >= warning_sec:
        return "warning"
    return "active"


# ---------------------------------------------------------------------------
# Project scanning (cached)
# ---------------------------------------------------------------------------
def _dir_signature(root: str) -> str:
    """Cheap signature: path + mtime + name for dirs up to depth 3."""
    parts: list[str] = []
    if not os.path.isdir(root):
        return ""
    try:
        for class_name in sorted(os.listdir(root)):
            class_path = os.path.join(root, class_name)
            if not os.path.isdir(class_path) or class_name.startswith("."):
                continue
            try:
                parts.append(f"{class_name}:{os.path.getmtime(class_path)}")
            except OSError:
                parts.append(class_name)
            for student_name in sorted(os.listdir(class_path)):
                student_path = os.path.join(class_path, student_name)
                if not os.path.isdir(student_path) or student_name.startswith("."):
                    continue
                try:
                    parts.append(f"{class_name}/{student_name}:{os.path.getmtime(student_path)}")
                except OSError:
                    parts.append(f"{class_name}/{student_name}")
                for project_name in sorted(os.listdir(student_path)):
                    project_path = os.path.join(student_path, project_name)
                    if not os.path.isdir(project_path) or project_name.startswith("."):
                        continue
                    try:
                        parts.append(
                            f"{class_name}/{student_name}/{project_name}:{os.path.getmtime(project_path)}"
                        )
                    except OSError:
                        parts.append(f"{class_name}/{student_name}/{project_name}")
    except OSError:
        return ""
    return "|".join(parts)


def _scan_projects_uncached() -> dict:
    """
    Scan projects/ into:
    { class_name: { student_name: [ {name, type, url}, ... ] } }
    Priority: python > html > apk > external
    """
    tree: dict = {}
    if not os.path.exists(PROJECTS_DIR):
        return tree

    for class_name in sorted(os.listdir(PROJECTS_DIR)):
        class_path = os.path.join(PROJECTS_DIR, class_name)
        if not os.path.isdir(class_path) or class_name.startswith("."):
            continue

        tree[class_name] = {}
        for student_name in sorted(os.listdir(class_path)):
            student_path = os.path.join(class_path, student_name)
            if not os.path.isdir(student_path) or student_name.startswith("."):
                continue

            projects = []
            for project_name in sorted(os.listdir(student_path)):
                project_dir = os.path.join(student_path, project_name)
                if not os.path.isdir(project_dir) or project_name.startswith("."):
                    continue

                main_py = os.path.join(project_dir, "main.py")
                index_html = os.path.join(project_dir, "index.html")
                link_txt = os.path.join(project_dir, "link.txt")

                apk_files = []
                try:
                    apk_files = [
                        f
                        for f in os.listdir(project_dir)
                        if f.endswith(".apk") and not f.startswith(".")
                    ]
                except OSError:
                    pass

                if os.path.exists(main_py):
                    projects.append({"name": project_name, "type": "python", "url": ""})
                elif os.path.exists(index_html):
                    projects.append({"name": project_name, "type": "html", "url": ""})
                elif apk_files:
                    projects.append({"name": project_name, "type": "apk", "url": apk_files[0]})
                elif os.path.exists(link_txt):
                    try:
                        with open(link_txt, "r", encoding="utf-8") as f:
                            url = f.read().strip()
                        if _safe_external_url(url):
                            projects.append({"name": project_name, "type": "external", "url": url})
                    except OSError:
                        pass

            if projects:
                for project in projects:
                    project["display_name"] = _project_display_name(class_name, student_name, project["name"])
                tree[class_name][student_name] = projects

        if not tree[class_name]:
            del tree[class_name]

    return tree


def scan_projects(force: bool = False) -> dict:
    """Return cached project tree; refresh when TTL expires or directory signature changes."""
    now = time.time()
    sig = _dir_signature(PROJECTS_DIR)
    cache_age = now - _project_cache["built_at"]
    if (
        not force
        and _project_cache["tree"] is not None
        and cache_age < PROJECT_CACHE_TTL_SEC
        and _project_cache["signature"] == sig
    ):
        return _project_cache["tree"]

    tree = _scan_projects_uncached()
    _project_cache["tree"] = tree
    _project_cache["built_at"] = now
    _project_cache["signature"] = sig
    return tree


def _project_display_name(class_name: str, student_name: str, project_name: str) -> str:
    if class_name == "DEMO" and student_name == "展示用學生":
        return re.sub(r"^\d+_", "", project_name)
    return project_name


def get_id(name: str) -> str:
    """Stable 10-char hex hash for URL tokens."""
    return hashlib.sha256(name.encode("utf-8")).hexdigest()[:10]


def decode_class_student(
    class_name_or_id: str, student_name_or_id: Optional[str] = None
) -> tuple[Optional[str], Optional[str]]:
    """Resolve hashed or raw class/student identifiers to folder names."""
    all_projects = scan_projects()

    if not student_name_or_id:
        for c in all_projects:
            for s in all_projects[c]:
                if get_id(f"{c}/{s}") == class_name_or_id:
                    return c, s

        if "/" in class_name_or_id:
            parts = class_name_or_id.split("/", 1)
            c, s = parts[0], parts[1]
            if c in all_projects and s in all_projects[c]:
                return c, s

        real_class = None
        for c in all_projects:
            if get_id(c) == class_name_or_id:
                real_class = c
                break
        if not real_class and class_name_or_id in all_projects:
            real_class = class_name_or_id
        return real_class, None

    real_class = None
    for c in all_projects:
        if get_id(c) == class_name_or_id:
            real_class = c
            break
    if not real_class and class_name_or_id in all_projects:
        real_class = class_name_or_id

    real_student = None
    if real_class and real_class in all_projects:
        for s in all_projects[real_class]:
            if get_id(f"{real_class}/{s}") == student_name_or_id:
                real_student = s
                break
        if not real_student and student_name_or_id in all_projects[real_class]:
            real_student = student_name_or_id

    return real_class, real_student


def _safe_project_path(*parts: str) -> Optional[str]:
    """Join under PROJECTS_DIR; return None if path escapes the root."""
    file_path = os.path.join(PROJECTS_DIR, *parts)
    abs_path = os.path.abspath(file_path)
    root = os.path.abspath(PROJECTS_DIR)
    if not abs_path.startswith(root + os.sep) and abs_path != root:
        return None
    return abs_path


def _resolve_python_project(class_name: str, student_name: str, project_name: str) -> str:
    """Resolve only catalogued Python projects beneath their student's folder."""
    student_projects = scan_projects().get(class_name, {}).get(student_name, [])
    allowed = any(item["name"] == project_name and item["type"] == "python" for item in student_projects)
    student_root = os.path.realpath(os.path.join(PROJECTS_DIR, class_name, student_name))
    project_root = os.path.realpath(os.path.join(student_root, project_name))
    if not allowed or os.path.commonpath((student_root, project_root)) != student_root:
        raise FileNotFoundError("project is not a listed Python project")
    main_file = os.path.join(project_root, "main.py")
    if not os.path.isfile(main_file):
        raise FileNotFoundError("project has no main.py")
    return project_root


def _safe_external_url(value: str) -> str:
    """Only expose absolute HTTPS project links to browser navigation/iframes."""
    value = value.strip()
    try:
        parsed = urlsplit(value)
        if parsed.scheme == "https" and parsed.hostname and not parsed.username and not parsed.password:
            return value
    except ValueError:
        pass
    return ""


def _stage_runner_project(source_root: str) -> tuple[str, str]:
    """Copy regular project files into an ephemeral, per-run writable directory."""
    session_root = tempfile.mkdtemp(prefix="run-", dir=RUNNER_TMP_DIR)
    os.chmod(session_root, 0o700)
    workspace = os.path.join(session_root, "project")
    os.mkdir(workspace, 0o700)
    total_bytes = 0
    total_files = 0
    ignored_dirs = {".git", ".venv", "venv", "__pycache__", "node_modules"}
    for current, dirs, files in os.walk(source_root, followlinks=False):
        dirs[:] = [name for name in dirs if name not in ignored_dirs and not os.path.islink(os.path.join(current, name))]
        relative = os.path.relpath(current, source_root)
        target_dir = workspace if relative == "." else os.path.join(workspace, relative)
        if target_dir != workspace:
            os.makedirs(target_dir, mode=0o700, exist_ok=True)
        for name in files:
            source = os.path.join(current, name)
            try:
                info = os.lstat(source)
            except OSError:
                continue
            if not stat.S_ISREG(info.st_mode):
                continue
            lowered = name.casefold()
            if lowered == ".env" or lowered.startswith(".env.") or lowered in {
                "credentials.json", "service-account.json", "secrets.json", "token.txt",
            }:
                continue
            total_files += 1
            total_bytes += info.st_size
            if total_files > RUNNER_PROJECT_FILE_LIMIT:
                shutil.rmtree(session_root, ignore_errors=True)
                raise ValueError(f"專案檔案數超過上限（{RUNNER_PROJECT_FILE_LIMIT} 個）")
            if total_bytes > RUNNER_PROJECT_LIMIT_MB * 1024 * 1024:
                shutil.rmtree(session_root, ignore_errors=True)
                raise ValueError(f"專案資料超過執行上限（{RUNNER_PROJECT_LIMIT_MB} MB）")
            destination = os.path.join(target_dir, name)
            shutil.copyfile(source, destination)
            os.chmod(destination, 0o600)
        os.chmod(target_dir, 0o700)
    if not os.path.isfile(os.path.join(workspace, "main.py")):
        shutil.rmtree(session_root, ignore_errors=True)
        raise FileNotFoundError("專案沒有可執行的 main.py")
    return session_root, workspace


def _runner_sandbox_command(workspace: str, socket_path: str) -> list[str]:
    """Build a fail-closed bubblewrap command with no network or app secrets."""
    if sys.platform != "linux" or not BWRAP_PATH or not PRLIMIT_PATH:
        raise RuntimeError("伺服器缺少 Linux bubblewrap/prlimit，已拒絕未隔離執行。")

    bwrap = [
        BWRAP_PATH,
        "--die-with-parent",
        "--new-session",
        "--unshare-user",
        "--unshare-pid",
        "--unshare-ipc",
        "--unshare-net",
        "--unshare-uts",
        "--unshare-cgroup-try",
        "--disable-userns",
    ]
    for system_path in ("/usr", "/lib", "/lib64", "/bin", "/sbin", "/etc", "/opt"):
        if os.path.exists(system_path):
            bwrap.extend(("--ro-bind", system_path, system_path))
    bwrap.extend((
        "--dev", "/dev",
        "--proc", "/proc",
        "--size", str(RUNNER_TMP_LIMIT_MB * 1024 * 1024),
        "--tmpfs", "/tmp",
        "--dir", "/workspace",
        "--dir", "/workspace/project",
        "--bind", workspace, "/workspace/project",
        "--ro-bind", os.path.join(BASE_DIR, "runner.py"), "/runner.py",
        "--ro-bind", MODELS_DIR, "/models",
        "--ro-bind", os.path.dirname(socket_path), "/run-ipc",
        "--clearenv",
        "--setenv", "PATH", "/usr/local/bin:/usr/bin:/bin",
        "--setenv", "HOME", "/tmp",
        "--setenv", "LANG", "C.UTF-8",
        "--setenv", "PYTHONNOUSERSITE", "1",
        "--setenv", "PYTHONDONTWRITEBYTECODE", "1",
        "--setenv", "SDL_VIDEODRIVER", "dummy",
        "--setenv", "SDL_AUDIODRIVER", "dummy",
        "--chdir", "/workspace/project",
        "--",
        sys.executable,
        "/runner.py",
        "--project_path", "/workspace/project/main.py",
        "--socket-path", "/run-ipc/runner.sock",
        "--models_dir", "/models",
    ))
    return _runner_with_resource_limits(bwrap)


def _runner_with_resource_limits(command: list[str]) -> list[str]:
    if not PRLIMIT_PATH:
        raise RuntimeError("伺服器缺少 prlimit，無法套用執行資源限制。")
    memory_bytes = RUNNER_MEMORY_LIMIT_MB * 1024 * 1024
    file_bytes = 128 * 1024 * 1024
    return [
        PRLIMIT_PATH,
        f"--cpu={RUNNER_CPU_LIMIT_SEC}:{RUNNER_CPU_LIMIT_SEC}",
        f"--as={memory_bytes}:{memory_bytes}",
        f"--fsize={file_bytes}:{file_bytes}",
        "--nproc=64:64",
        "--nofile=128:128",
        "--core=0:0",
        "--",
        *command,
    ]


def _runner_command(workspace: str, socket_path: str) -> list[str]:
    if RUNNER_ISOLATION_MODE == "sandbox":
        return _runner_sandbox_command(workspace, socket_path)
    if RUNNER_ISOLATION_MODE != "process" or sys.platform != "linux":
        raise RuntimeError("不支援的 Python 執行模式。")
    return _runner_with_resource_limits([
        sys.executable,
        os.path.join(BASE_DIR, "runner.py"),
        "--project_path", os.path.join(workspace, "main.py"),
        "--socket-path", socket_path,
        "--models_dir", MODELS_DIR,
    ])


def _runner_environment(workspace: str) -> dict[str, str]:
    """Do not pass application tokens or admin credentials to child processes."""
    return {
        "PATH": "/usr/local/bin:/usr/bin:/bin",
        "HOME": workspace,
        "LANG": "C.UTF-8",
        "PYTHONNOUSERSITE": "1",
        "PYTHONDONTWRITEBYTECODE": "1",
        "PYTHONUNBUFFERED": "1",
        "SDL_VIDEODRIVER": "dummy",
        "SDL_AUDIODRIVER": "dummy",
        "OPENBLAS_NUM_THREADS": "1",
        "OMP_NUM_THREADS": "1",
    }


# ---------------------------------------------------------------------------
# Static / logo
# ---------------------------------------------------------------------------
if os.path.isdir(STATIC_DIR):
    app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")


@app.get("/images/logo-light.png")
@app.get("/logo.png")
async def get_logo():
    return FileResponse(os.path.join(TEMPLATES_DIR, "logo-light.png"))


# Controlled project file serving only (no raw StaticFiles mount on projects/)
@app.get("/static_projects/{student_token}/{project_name}/{path:path}")
async def serve_static_project_file(student_token: str, project_name: str, path: str):
    real_class, real_student = decode_class_student(student_token)

    if not real_class or not real_student:
        # Legacy: /static_projects/{class_id}/{student_id}/{project}/{path}
        parts = path.split("/", 1)
        if len(parts) == 2:
            actual_project_name, actual_path = parts[0], parts[1]
            real_class, real_student = decode_class_student(student_token, project_name)
            if real_class and real_student:
                file_path = _safe_project_path(
                    real_class, real_student, actual_project_name, actual_path
                )
                if file_path and os.path.isfile(file_path):
                    return FileResponse(file_path)
        return HTMLResponse("Not Found", status_code=404)

    file_path = _safe_project_path(real_class, real_student, project_name, path)
    if not file_path or not os.path.isfile(file_path):
        return HTMLResponse("File Not Found", status_code=404)
    return FileResponse(file_path)


# ---------------------------------------------------------------------------
# HTTP routes
# ---------------------------------------------------------------------------
@app.get("/health")
async def health():
    return {
        "status": "ok",
        "active_sessions": len(active_sessions),
        "max_concurrent_runners": MAX_CONCURRENT_RUNNERS,
        "session_timeout_sec": SESSION_TIMEOUT_SEC,
        "idle_warning_sec": IDLE_WARNING_SEC,
        "idle_timeout_sec": IDLE_TIMEOUT_SEC,
        "heartbeat_timeout_sec": HEARTBEAT_TIMEOUT_SEC,
    }


@app.get("/api/metrics")
async def metrics():
    return {
        "active_sessions": len(active_sessions),
        "max_concurrent_runners": MAX_CONCURRENT_RUNNERS,
        "sessions": list(active_sessions.values()),
        **_metrics,
    }


@app.get("/", response_class=HTMLResponse)
async def read_index(request: Request):
    return templates.TemplateResponse(
        request=request,
        name="index.html",
        context={"restrict_class": "", "restrict_student": ""},
    )


@app.get("/class/{class_name}", response_class=HTMLResponse)
async def read_class(request: Request, class_name: str):
    real_class, _ = decode_class_student(class_name)
    if not real_class:
        return HTMLResponse("Class not found", status_code=404)
    return templates.TemplateResponse(
        request=request,
        name="index.html",
        context={"restrict_class": real_class, "restrict_student": ""},
    )


@app.get("/student/{student_token}", response_class=HTMLResponse)
async def read_student(request: Request, student_token: str):
    real_class, real_student = decode_class_student(student_token)
    if not real_class or not real_student:
        return HTMLResponse("Student not found", status_code=404)
    return templates.TemplateResponse(
        request=request,
        name="index.html",
        context={
            "restrict_class": real_class,
            "restrict_student": real_student,
        },
    )


@app.get("/student/{class_name}/{student_name}", response_class=HTMLResponse)
async def read_student_old(request: Request, class_name: str, student_name: str):
    real_class, real_student = decode_class_student(class_name, student_name)
    if not real_class or not real_student:
        return HTMLResponse("Student not found", status_code=404)
    student_token = get_id(f"{real_class}/{real_student}")
    return RedirectResponse(url=f"/student/{student_token}", status_code=307)


@app.get("/api/projects", response_class=JSONResponse)
async def get_projects(class_name: str = None, student_name: str = None):
    all_projects = scan_projects()
    class_ids = {c: get_id(c) for c in all_projects}
    student_ids = {
        f"{c}/{s}": get_id(f"{c}/{s}") for c in all_projects for s in all_projects[c]
    }

    if class_name:
        real_class, real_student = decode_class_student(class_name, student_name)
        if real_class in all_projects:
            if real_student:
                if real_student in all_projects[real_class]:
                    filtered = {real_class: {real_student: all_projects[real_class][real_student]}}
                else:
                    filtered = {real_class: {}}
            else:
                filtered = {real_class: all_projects[real_class]}
        else:
            filtered = {}
    else:
        filtered = all_projects

    return {"projects": filtered, "class_ids": class_ids, "student_ids": student_ids}


@app.get("/run/{student_token}/{project_name}", response_class=HTMLResponse)
async def run_project_page(request: Request, student_token: str, project_name: str):
    real_class, real_student = decode_class_student(student_token)
    if not real_class or not real_student:
        return HTMLResponse("Project not found", status_code=404)

    proj_dir = os.path.join(PROJECTS_DIR, real_class, real_student, project_name)
    proj_type = "unknown"
    external_url = ""
    apk_name = ""
    apk_size = ""
    appetize_key = ""
    view_only = request.query_params.get("view") == "1"

    apk_files = []
    if os.path.isdir(proj_dir):
        try:
            apk_files = [
                f for f in os.listdir(proj_dir) if f.endswith(".apk") and not f.startswith(".")
            ]
        except OSError:
            pass

    if os.path.exists(os.path.join(proj_dir, "main.py")):
        proj_type = "python"
    elif os.path.exists(os.path.join(proj_dir, "index.html")):
        proj_type = "html"
    elif apk_files:
        proj_type = "apk"
        apk_name = apk_files[0]
        try:
            size_bytes = os.path.getsize(os.path.join(proj_dir, apk_name))
            if size_bytes < 1024 * 1024:
                apk_size = f"{size_bytes / 1024:.1f} KB"
            else:
                apk_size = f"{size_bytes / (1024 * 1024):.1f} MB"
        except OSError:
            pass
        appetize_path = os.path.join(proj_dir, "appetize.txt")
        if os.path.exists(appetize_path):
            try:
                with open(appetize_path, "r", encoding="utf-8") as f:
                    candidate_key = f.read().strip()
                    if re.fullmatch(r"[A-Za-z0-9_-]{1,256}", candidate_key):
                        appetize_key = candidate_key
            except OSError:
                pass
    elif os.path.exists(os.path.join(proj_dir, "link.txt")):
        try:
            with open(os.path.join(proj_dir, "link.txt"), "r", encoding="utf-8") as f:
                external_url = _safe_external_url(f.read())
            if external_url:
                proj_type = "external"
        except OSError:
            pass

    return templates.TemplateResponse(
        request=request,
        name="player.html",
        context={
            "class_name": real_class,
            "student_name": real_student,
            "student_token": student_token,
            "project_name": project_name,
            "project_display_name": _project_display_name(real_class, real_student, project_name),
            "project_type": proj_type,
            "external_url": external_url,
            "apk_name": apk_name,
            "apk_size": apk_size,
            "appetize_key": appetize_key,
            "view_only": view_only,
            "session_timeout_sec": SESSION_TIMEOUT_SEC,
            "idle_warning_sec": IDLE_WARNING_SEC,
            "idle_timeout_sec": IDLE_TIMEOUT_SEC,
            "heartbeat_interval_sec": min(20, max(10, HEARTBEAT_TIMEOUT_SEC // 3)),
            "max_concurrent_runners": MAX_CONCURRENT_RUNNERS,
        },
    )


@app.get("/run/{class_name}/{student_name}/{project_name}", response_class=HTMLResponse)
async def run_project_page_old(
    request: Request, class_name: str, student_name: str, project_name: str
):
    real_class, real_student = decode_class_student(class_name, student_name)
    if not real_class or not real_student:
        return HTMLResponse("Project not found", status_code=404)
    student_token = get_id(f"{real_class}/{real_student}")
    qs = f"?{request.url.query}" if request.url.query else ""
    return RedirectResponse(
        url=f"/run/{student_token}/{project_name}{qs}", status_code=307
    )


# ---------------------------------------------------------------------------
# Process helpers
# ---------------------------------------------------------------------------
def _kill_process_group(process: asyncio.subprocess.Process) -> None:
    """Terminate the entire process group (runner + children)."""
    if process.returncode is not None:
        return
    pid = process.pid
    try:
        if sys.platform == "win32":
            process.terminate()
        else:
            try:
                os.killpg(pid, signal.SIGTERM)
            except (ProcessLookupError, PermissionError, OSError):
                process.terminate()
    except ProcessLookupError:
        pass


async def _cleanup_process(process: asyncio.subprocess.Process) -> None:
    if process is None or process.returncode is not None:
        return
    log.info("Cleaning up process group pid=%s", process.pid)
    _kill_process_group(process)
    try:
        await asyncio.wait_for(process.wait(), timeout=2.0)
    except asyncio.TimeoutError:
        try:
            if sys.platform == "win32":
                process.kill()
            else:
                try:
                    os.killpg(process.pid, signal.SIGKILL)
                except (ProcessLookupError, PermissionError, OSError):
                    process.kill()
        except ProcessLookupError:
            pass
        try:
            await process.wait()
        except Exception:
            pass


async def _capture_bounded_stderr(
    stream: asyncio.StreamReader, captured: bytearray, limit: int = 4096
) -> None:
    """Drain child stderr without allowing an unbounded buffer or blocked pipe."""
    while True:
        chunk = await stream.read(1024)
        if not chunk:
            return
        remaining = limit - len(captured)
        if remaining > 0:
            captured.extend(chunk[:remaining])


# ---------------------------------------------------------------------------
# WebSocket play session
# ---------------------------------------------------------------------------
async def play_project_ws_internal(
    websocket: WebSocket, real_class: str, real_student: str, project_name: str
):
    if not real_class or not real_student:
        await websocket.accept()
        await websocket.send_json(
            {"type": "log", "data": "錯誤: 找不到該專案的執行路徑！\n"}
        )
        await websocket.close()
        return

    try:
        source_root = _resolve_python_project(real_class, real_student, project_name)
    except (OSError, ValueError):
        await websocket.accept()
        await websocket.send_json(
            {"type": "log", "data": "錯誤: 找不到該專案的 main.py 檔案！\n"}
        )
        await websocket.close()
        return

    # Concurrency gate: reject immediately if all runner slots are busy
    acquired = False
    try:
        await asyncio.wait_for(runner_semaphore.acquire(), timeout=0.05)
        acquired = True
    except asyncio.TimeoutError:
        _metrics["sessions_rejected"] += 1
        await websocket.accept()
        await websocket.send_json(
            {
                "type": "log",
                "data": (
                    f"目前展示名額已滿（最多同時 {MAX_CONCURRENT_RUNNERS} 個專案）。"
                    "請稍後再試。\n"
                ),
            }
        )
        await websocket.send_json({"type": "status", "data": "busy"})
        await websocket.close(code=1013)
        return

    remote_ip = websocket.client.host if websocket.client else "unknown"
    if not allow_runner_start(remote_ip):
        runner_semaphore.release()
        acquired = False
        _metrics["sessions_rejected"] += 1
        await websocket.accept()
        await websocket.send_json({"type": "log", "data": "這個網路來源的專案啟動次數已達上限，請稍後再試。\n"})
        await websocket.send_json({"type": "status", "data": "rate_limited"})
        await websocket.close(code=4429, reason="runner start rate limited")
        return

    await websocket.accept()
    session_id = f"{real_class}/{real_student}/{project_name}/{time.time():.0f}"
    started_at = time.time()
    started_monotonic = time.monotonic()
    last_activity_at = started_monotonic
    last_client_seen_at = started_monotonic

    async with _sessions_lock:
        active_sessions[session_id] = {
            "class": real_class,
            "student": real_student,
            "project": project_name,
            "started_at": started_at,
            "last_activity_at": started_at,
        }
        _metrics["sessions_started"] += 1

    process = None
    process_stderr_task = None
    process_stderr = bytearray()
    ipc_server = None
    runner_writer = None
    session_root = None

    try:
        connected_event = asyncio.Event()
        runner_reader = None

        async def handle_runner_conn(reader, writer):
            nonlocal runner_reader, runner_writer
            runner_reader = reader
            runner_writer = writer
            connected_event.set()

        try:
            session_root, workspace = _stage_runner_project(source_root)
            ipc_dir = os.path.join(session_root, "ipc")
            os.mkdir(ipc_dir, 0o700)
            socket_path = os.path.join(ipc_dir, "runner.sock")
            cmd = _runner_command(workspace, socket_path)
            ipc_server = await asyncio.start_unix_server(
                handle_runner_conn, path=socket_path
            )
            os.chmod(socket_path, 0o600)
        except (OSError, RuntimeError, ValueError) as exc:
            log.warning("Unable to prepare runner: %s", exc)
            await websocket.send_json({"type": "log", "data": f"專案執行器無法啟動，已停止執行：{exc}\n"})
            await websocket.send_json({"type": "status", "data": "error"})
            return

        popen_kwargs: dict[str, Any] = {
            "stdout": subprocess.DEVNULL,
            "stderr": subprocess.PIPE,
            "cwd": workspace,
            "env": _runner_environment(workspace),
        }
        if sys.platform != "win32":
            popen_kwargs["start_new_session"] = True

        process = await asyncio.create_subprocess_exec(*cmd, **popen_kwargs)
        if process.stderr is not None:
            process_stderr_task = asyncio.create_task(
                _capture_bounded_stderr(process.stderr, process_stderr)
            )
        log.info("Started runner mode=%s pid=%s session=%s", RUNNER_ISOLATION_MODE, process.pid, session_id)

        try:
            await asyncio.wait_for(
                connected_event.wait(), timeout=RUNNER_CONNECT_TIMEOUT_SEC
            )
            log.info("Runner pid=%s connected via private IPC", process.pid)
            await websocket.send_json(
                {"type": "status", "data": "running"}
            )
            await websocket.send_json(
                {
                    "type": "log",
                    "data": f"[系統] 專案執行器已就緒（最長 {SESSION_TIMEOUT_SEC // 60} 分鐘）。\n",
                }
            )
        except asyncio.TimeoutError:
            _metrics["sessions_failed_connect"] += 1
            if process.returncode is None:
                try:
                    await asyncio.wait_for(process.wait(), timeout=0.2)
                except asyncio.TimeoutError:
                    pass
            if process.returncode is not None and process_stderr_task is not None:
                await asyncio.gather(process_stderr_task, return_exceptions=True)
            stderr_detail = process_stderr.decode("utf-8", errors="replace").strip()
            log.error(
                "Runner pid=%s IPC connect timeout; returncode=%s; stderr=%s",
                process.pid,
                process.returncode,
                stderr_detail or "<empty>",
            )
            await websocket.send_json(
                {"type": "log", "data": (
                    "安全執行環境無法連線（可能是主機未啟用 Linux namespace）；已拒絕未隔離執行。\n"
                    if RUNNER_ISOLATION_MODE == "sandbox"
                    else "專案執行器啟動失敗，請稍後重試或請老師查看伺服器日誌。\n"
                )}
            )
            await websocket.send_json({"type": "status", "data": "error"})
            return

        async def tcp_to_ws():
            try:
                while True:
                    type_bytes = await runner_reader.readexactly(4)
                    msg_type = type_bytes.decode("utf-8")
                    len_bytes = await runner_reader.readexactly(4)
                    length = int.from_bytes(len_bytes, byteorder="big")
                    payload = await runner_reader.readexactly(length)

                    if msg_type == "FRME":
                        await websocket.send_bytes(payload)
                    elif msg_type == "LOGS":
                        log_text = payload.decode("utf-8", errors="ignore")
                        await websocket.send_json({"type": "log", "data": log_text})
                    elif msg_type == "EXIT":
                        log.info("Runner pid=%s sent EXIT", process.pid)
                        await websocket.send_json({"type": "status", "data": "stopped"})
                        break
            except asyncio.IncompleteReadError:
                log.info("Runner pid=%s TCP incomplete read", process.pid)
            except WebSocketDisconnect:
                pass
            except Exception as e:
                log.exception("tcp_to_ws error: %s", e)

        async def ws_to_tcp():
            nonlocal last_activity_at, last_client_seen_at
            try:
                while True:
                    data = await websocket.receive()
                    now = time.monotonic()
                    last_client_seen_at = now
                    if "text" in data:
                        message = json.loads(data["text"])
                        mtype = message.get("type")

                        if mtype == "heartbeat":
                            continue

                        if mtype == "activity":
                            last_activity_at = now
                            active_sessions[session_id]["last_activity_at"] = time.time()
                        elif mtype in ("keydown", "keyup"):
                            last_activity_at = now
                            active_sessions[session_id]["last_activity_at"] = time.time()
                            payload = json.dumps(message).encode("utf-8")
                            runner_writer.write(
                                b"KEYS" + len(payload).to_bytes(4, "big") + payload
                            )
                            await runner_writer.drain()
                        elif mtype in ("mousemove", "mousedown", "mouseup"):
                            last_activity_at = now
                            active_sessions[session_id]["last_activity_at"] = time.time()
                            payload = json.dumps(message).encode("utf-8")
                            runner_writer.write(
                                b"MOUS" + len(payload).to_bytes(4, "big") + payload
                            )
                            await runner_writer.drain()
                        elif mtype == "cv2key":
                            last_activity_at = now
                            active_sessions[session_id]["last_activity_at"] = time.time()
                            payload = json.dumps(
                                {"key_code": message.get("keyCode")}
                            ).encode("utf-8")
                            runner_writer.write(
                                b"CV2K" + len(payload).to_bytes(4, "big") + payload
                            )
                            await runner_writer.drain()
                        elif mtype == "terminal_input":
                            last_activity_at = now
                            active_sessions[session_id]["last_activity_at"] = time.time()
                            text = message.get("data", "")
                            if not text.endswith("\n"):
                                text += "\n"
                            payload = text.encode("utf-8")
                            runner_writer.write(
                                b"INPT" + len(payload).to_bytes(4, "big") + payload
                            )
                            await runner_writer.drain()
                        elif mtype == "pygame_quit":
                            last_activity_at = now
                            runner_writer.write(b"QUIT" + (0).to_bytes(4, "big") + b"")
                            await runner_writer.drain()
                    elif "bytes" in data:
                        payload = data["bytes"]
                        runner_writer.write(
                            b"CAMF" + len(payload).to_bytes(4, "big") + payload
                        )
                        await runner_writer.drain()
            except WebSocketDisconnect:
                log.info("Browser WebSocket disconnected session=%s", session_id)
            except Exception as e:
                log.exception("ws_to_tcp error: %s", e)

        async def session_watchdog():
            warning_sent = False
            while True:
                await asyncio.sleep(1)
                now = time.monotonic()
                idle_for = now - last_activity_at
                session_age = now - started_monotonic
                heartbeat_age = now - last_client_seen_at
                idle_state = _session_idle_state(idle_for)

                if session_age >= SESSION_TIMEOUT_SEC:
                    _metrics["sessions_timed_out"] += 1
                    log.warning("Session hard timeout session=%s", session_id)
                    await websocket.send_json(
                        {
                            "type": "log",
                            "data": f"\n[系統] 已達最長執行時間（{SESSION_TIMEOUT_SEC // 60} 分鐘），自動結束。\n",
                        }
                    )
                    await websocket.send_json({"type": "status", "data": "timeout"})
                    break

                if heartbeat_age >= HEARTBEAT_TIMEOUT_SEC:
                    _metrics["sessions_heartbeat_timed_out"] += 1
                    log.warning("Session heartbeat timeout session=%s", session_id)
                    await websocket.send_json(
                        {
                            "type": "log",
                            "data": "\n[系統] 瀏覽器已離線，自動釋放展示名額。\n",
                        }
                    )
                    await websocket.send_json(
                        {"type": "status", "data": "connection_timeout"}
                    )
                    break

                if idle_state == "timeout":
                    _metrics["sessions_idle_timed_out"] += 1
                    log.info("Session idle timeout session=%s", session_id)
                    await websocket.send_json(
                        {
                            "type": "log",
                            "data": f"\n[系統] 已閒置 {IDLE_TIMEOUT_SEC // 60} 分鐘，自動結束並釋放展示名額。\n",
                        }
                    )
                    await websocket.send_json(
                        {"type": "status", "data": "idle_timeout"}
                    )
                    break

                if idle_state == "warning" and not warning_sent:
                    warning_sent = True
                    await websocket.send_json(
                        {
                            "type": "idle_warning",
                            "remaining_sec": max(
                                0, int(IDLE_TIMEOUT_SEC - idle_for)
                            ),
                        }
                    )
                elif idle_state == "active" and warning_sent:
                    warning_sent = False
                    await websocket.send_json({"type": "idle_resumed"})

            try:
                await websocket.close(code=1000)
            except Exception:
                pass

        tcp_task = asyncio.create_task(tcp_to_ws())
        ws_task = asyncio.create_task(ws_to_tcp())
        watchdog = asyncio.create_task(session_watchdog())
        try:
            _, pending = await asyncio.wait(
                {tcp_task, ws_task, watchdog},
                return_when=asyncio.FIRST_COMPLETED,
            )
            for task in pending:
                task.cancel()
            await asyncio.gather(*pending, return_exceptions=True)
        finally:
            for task in (tcp_task, ws_task, watchdog):
                if not task.done():
                    task.cancel()
            await asyncio.gather(
                tcp_task, ws_task, watchdog, return_exceptions=True
            )

    finally:
        if process is not None:
            await _cleanup_process(process)
        if process_stderr_task is not None and not process_stderr_task.done():
            try:
                await asyncio.wait_for(process_stderr_task, timeout=1.0)
            except asyncio.TimeoutError:
                process_stderr_task.cancel()
                await asyncio.gather(process_stderr_task, return_exceptions=True)
        if runner_writer is not None:
            try:
                runner_writer.close()
                await runner_writer.wait_closed()
            except Exception:
                pass
        if ipc_server is not None:
            ipc_server.close()
            await ipc_server.wait_closed()
        if session_root is not None:
            shutil.rmtree(session_root, ignore_errors=True)
        try:
            await websocket.close()
        except Exception:
            pass
        async with _sessions_lock:
            active_sessions.pop(session_id, None)
        if acquired:
            runner_semaphore.release()
        log.info(
            "Session ended session=%s duration=%.1fs",
            session_id,
            time.time() - started_at,
        )


@app.websocket("/ws/play/{student_token}/{project_name}")
async def play_project_ws(websocket: WebSocket, student_token: str, project_name: str):
    real_class, real_student = decode_class_student(student_token)
    await play_project_ws_internal(websocket, real_class, real_student, project_name)


@app.websocket("/ws/play/{class_name}/{student_name}/{project_name}")
async def play_project_ws_old(
    websocket: WebSocket, class_name: str, student_name: str, project_name: str
):
    real_class, real_student = decode_class_student(class_name, student_name)
    await play_project_ws_internal(websocket, real_class, real_student, project_name)


# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    import uvicorn

    parser = argparse.ArgumentParser(description="Telearn Arcade Hub")
    parser.add_argument("--host", default="127.0.0.1", help="Server host")
    parser.add_argument("--port", type=int, default=8000, help="Server port")
    parser.add_argument(
        "--reload",
        action="store_true",
        default=os.environ.get("DEV_RELOAD", "").lower() in ("1", "true", "yes"),
        help="Enable auto-reload (dev only)",
    )
    args = parser.parse_args()

    uvicorn.run(
        "server:app",
        host=args.host,
        port=args.port,
        reload=args.reload,
    )