Spaces:

Telearn
/

Project_Demo

Running

Add content admin and interactive Python console
ed832a7 2 months ago
36.1 kB
"""
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 subprocess
import sys
import time
from typing import Any, Optional

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"))

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)

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):
    """Enable SharedArrayBuffer for interactive input inside the Python worker."""
    response = await call_next(request)
    response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
    response.headers["Cross-Origin-Embedder-Policy"] = "credentialless"
    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}


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 url:
                            projects.append({"name": project_name, "type": "external", "url": url})
                    except OSError:
                        pass

            if projects:
                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 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


# ---------------------------------------------------------------------------
# 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(
        "index.html",
        {"request": request, "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(
        "index.html",
        {"request": request, "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(
        "index.html",
        {
            "request": request,
            "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:
                    appetize_key = f.read().strip()
            except OSError:
                pass
    elif os.path.exists(os.path.join(proj_dir, "link.txt")):
        proj_type = "external"
        try:
            with open(os.path.join(proj_dir, "link.txt"), "r", encoding="utf-8") as f:
                external_url = f.read().strip()
        except OSError:
            pass

    return templates.TemplateResponse(
        "player.html",
        {
            "request": request,
            "class_name": real_class,
            "student_name": real_student,
            "student_token": student_token,
            "project_name": 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


# ---------------------------------------------------------------------------
# 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

    project_path = os.path.join(
        PROJECTS_DIR, real_class, real_student, project_name, "main.py"
    )
    if not os.path.exists(project_path):
        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

    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
    tcp_server = None
    runner_writer = 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()

        tcp_server = await asyncio.start_server(handle_runner_conn, "127.0.0.1", 0)
        tcp_port = tcp_server.sockets[0].getsockname()[1]

        runner_script = os.path.join(BASE_DIR, "runner.py")
        cmd = [
            sys.executable,
            runner_script,
            "--project_path",
            project_path,
            "--port",
            str(tcp_port),
            "--models_dir",
            MODELS_DIR,
        ]

        popen_kwargs: dict[str, Any] = {
            "stdout": subprocess.DEVNULL,
            "stderr": subprocess.DEVNULL,
        }
        if sys.platform != "win32":
            popen_kwargs["start_new_session"] = True

        process = await asyncio.create_subprocess_exec(*cmd, **popen_kwargs)
        log.info(
            "Started runner pid=%s session=%s port=%s",
            process.pid,
            session_id,
            tcp_port,
        )

        try:
            await asyncio.wait_for(
                connected_event.wait(), timeout=RUNNER_CONNECT_TIMEOUT_SEC
            )
            log.info("Runner pid=%s connected via TCP", 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
            log.error("Runner pid=%s connect timeout", process.pid)
            await websocket.send_json(
                {"type": "log", "data": "伺服器錯誤: 執行模擬器連線超時！\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 runner_writer is not None:
            try:
                runner_writer.close()
                await runner_writer.wait_closed()
            except Exception:
                pass
        if tcp_server is not None:
            tcp_server.close()
            await tcp_server.wait_closed()
        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,
    )