#!/usr/bin/env python3
"""
OpenVoiceUI Server — Entry Point

Initialises the Flask application and registers all route blueprints.
Routes are split into focused blueprints under routes/; this file handles
startup wiring, session management, usage tracking, and standalone endpoints
that don't belong to a specific feature blueprint.

Start:
    venv/bin/python3 server.py

See README.md for full setup instructions.
"""

import asyncio
import base64
import faulthandler
import json
import logging
import os
import queue
import re
import requests
import shutil
import signal
import sqlite3
import subprocess
import tempfile
import threading
import time
import uuid
from datetime import datetime
from pathlib import Path

import websockets
from dotenv import load_dotenv
from flask import Response, g, request, jsonify

faulthandler.enable()  # print traceback on hard crashes (SIGSEGV etc.)

# Load environment variables before anything else
env_path = Path(__file__).parent / ".env"
load_dotenv(env_path, override=True)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

SERVER_START_TIME = time.time()


# ---------------------------------------------------------------------------
# Faster-Whisper — lazy-loaded on first /api/stt/local request
# ---------------------------------------------------------------------------

_whisper_model = None


def get_whisper_model():
    """LEGACY in-process Whisper. Only used if the shared STT service is absent.

    Prefer the shared jambot-parakeet service (see local_stt below): the model is
    ~680MB and baking it into every one of 24+ OVU containers would be wasteful
    on a memory-constrained box. faster-whisper is also NOT installed by default
    (commented out in requirements.txt), so this path normally raises.
    """
    global _whisper_model
    if _whisper_model is None:
        try:
            from faster_whisper import WhisperModel
        except ImportError:
            raise ImportError(
                "Local STT requires either the shared jambot-parakeet service "
                "(preferred — set PARAKEET_STT_URL) or faster-whisper installed "
                "in this container."
            )
        logger.info("Loading Faster-Whisper model (first STT request)...")
        _whisper_model = WhisperModel("tiny", device="cpu", compute_type="float32")
        logger.info("Faster-Whisper model ready.")
    return _whisper_model


# Shared on-box STT (NVIDIA Parakeet TDT 0.6B v3) on the jambot-shared network.
# Free, keyless, audio never leaves the server. See deploy/parakeet-stt/.
PARAKEET_STT_URL = os.getenv("PARAKEET_STT_URL", "http://jambot-parakeet:8770")


def _parakeet_transcribe(audio_bytes, filename="audio.webm", timeout=180):
    """POST to the shared STT service. Returns transcript str, or None if the
    service is unreachable so the caller can fall back rather than fail."""
    import requests
    try:
        r = requests.post(
            f"{PARAKEET_STT_URL}/transcribe",
            files={"audio": (filename, audio_bytes)},
            timeout=timeout,
        )
        if r.status_code != 200:
            logger.warning("parakeet STT http %s: %s", r.status_code, r.text[:200])
            return None
        return (r.json() or {}).get("text", "")
    except Exception as e:
        # Unreachable is a fallback condition, not an error to surface upward.
        logger.info("parakeet STT unavailable (%s) — falling back", e)
        return None


# ---------------------------------------------------------------------------
# Flask app factory + blueprint registration
# ---------------------------------------------------------------------------

from app import create_app
app, sock = create_app()

from routes.music import music_bp
app.register_blueprint(music_bp)

from routes.canvas import (
    canvas_bp,
    canvas_context,
    update_canvas_context,
    extract_canvas_page_content,
    get_canvas_context,
    load_canvas_manifest,
    save_canvas_manifest,
    add_page_to_manifest,
    sync_canvas_manifest,
    CANVAS_MANIFEST_PATH,
    CANVAS_PAGES_DIR,
    CATEGORY_ICONS,
    CATEGORY_COLORS,
)
app.register_blueprint(canvas_bp)

# Seed default pages into canvas-pages on startup (ships with the app image)
# Default pages are app infrastructure (e.g. desktop menu) — auth is skipped in canvas.py.
# Pages with a <!-- openvoiceui-version: X --> comment are re-seeded when version changes.
from services.paths import DEFAULT_PAGES_DIR

_VERSION_RE = re.compile(r'<!--\s*openvoiceui-version:\s*(\S+)\s*-->')

def _extract_page_version(path, max_lines=5):
    """Read version stamp from the first few lines of an HTML file."""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            for _, line in zip(range(max_lines), f):
                m = _VERSION_RE.search(line)
                if m:
                    return m.group(1)
    except (OSError, UnicodeDecodeError):
        pass
    return None

if DEFAULT_PAGES_DIR.is_dir():
    CANVAS_PAGES_DIR.mkdir(parents=True, exist_ok=True)
    for src in DEFAULT_PAGES_DIR.iterdir():
        if not src.is_file():
            continue
        dest = CANVAS_PAGES_DIR / src.name
        # A single non-writable tenant page (e.g. a canvas-pages copy owned by a
        # different uid with restrictive perms) must NOT crash the whole server boot.
        # Log-and-skip so seeding stays best-effort. (host 2026-06-28 — fleet-roll fix:
        # tenant-k crash-looped here when website-creator.html was uid-1000/664 vs appuser 1001.)
        try:
            if not dest.exists():
                shutil.copy2(src, dest)
                logger.info("Seeded default page: %s", src.name)
            else:
                # Re-seed if the shipped version is newer than the runtime copy
                src_ver = _extract_page_version(src)
                if src_ver is not None:
                    dest_ver = _extract_page_version(dest)
                    if dest_ver != src_ver:
                        shutil.copy2(src, dest)
                        logger.info(
                            "Re-seeded default page %s (version %s -> %s)",
                            src.name, dest_ver, src_ver,
                        )
        except (PermissionError, OSError) as _seed_exc:
            logger.warning("Skipped seeding default page %s (non-fatal): %s", src.name, _seed_exc)

# Seed shared canvas modules (canvas-shared/*) into canvas-pages/_shared/ so pages can import
# them from /pages/_shared/<file> (e.g. html-surfaces.js for the bundled threejs-html-surfaces skill).
from services.paths import CANVAS_SHARED_DIR
from services.canvas_shared import seed_canvas_shared
seed_canvas_shared(CANVAS_SHARED_DIR, CANVAS_PAGES_DIR)

from routes.static_files import static_files_bp, DJ_SOUNDS, SOUNDS_DIR
app.register_blueprint(static_files_bp)

from routes.intake import intake_bp
app.register_blueprint(intake_bp)

from routes.albums import albums_bp
app.register_blueprint(albums_bp)

from routes.admin import admin_bp
app.register_blueprint(admin_bp)

from routes.theme import theme_bp
app.register_blueprint(theme_bp)

from routes.canvas_styles import canvas_styles_bp
app.register_blueprint(canvas_styles_bp)

from routes.conversation import conversation_bp, clean_for_tts
app.register_blueprint(conversation_bp)

from routes.profiles import profiles_bp
app.register_blueprint(profiles_bp)

from routes.elevenlabs_hybrid import elevenlabs_hybrid_bp
app.register_blueprint(elevenlabs_hybrid_bp)

from routes.instructions import instructions_bp
app.register_blueprint(instructions_bp)

from routes.greetings import greetings_bp
app.register_blueprint(greetings_bp)

from routes.suno import suno_bp
app.register_blueprint(suno_bp)

from routes.fal import fal_bp
app.register_blueprint(fal_bp)

from routes.story import story_bp
app.register_blueprint(story_bp)

from routes.airadio_bridge import airadio_bp
app.register_blueprint(airadio_bp)

# Voice Agent Tool Bus — /api/tools/* (catalog, invoke, jobs, SSE events).
# Spec: docs/jambot/voice-agent-tool-bus.md
from routes.tools import tools_bp
app.register_blueprint(tools_bp)

try:
    from routes.song_tagger import song_tagger_bp
    app.register_blueprint(song_tagger_bp)
except ImportError:
    pass

from routes.vision import vision_bp
app.register_blueprint(vision_bp)

from routes.transcripts import transcripts_bp
app.register_blueprint(transcripts_bp)

from routes.audio_clips import audio_clips_bp
app.register_blueprint(audio_clips_bp)

from routes.pi import pi_bp
app.register_blueprint(pi_bp)

from routes.onboarding import onboarding_bp
app.register_blueprint(onboarding_bp)

from routes.image_gen import image_gen_bp
app.register_blueprint(image_gen_bp)
from routes.grid_gen import grid_gen_bp
app.register_blueprint(grid_gen_bp)

from routes.meshy import meshy_bp
app.register_blueprint(meshy_bp)

from routes.chat import chat_bp
app.register_blueprint(chat_bp)

from routes.workspace import workspace_bp
app.register_blueprint(workspace_bp)

from routes.icons import icons_bp
from routes.report_issue import report_issue_bp
app.register_blueprint(icons_bp)
app.register_blueprint(report_issue_bp)

from routes.registry import registry_bp
app.register_blueprint(registry_bp)

from routes.chatgpt_import import chatgpt_import_bp
app.register_blueprint(chatgpt_import_bp)

# Plugin system — auto-discover and load installed plugins
from routes.plugins import plugins_bp
app.register_blueprint(plugins_bp)

from routes.custom_faces import custom_faces_bp
app.register_blueprint(custom_faces_bp)

from routes.vault import vault_bp
app.register_blueprint(vault_bp)

from routes.identity import identity_bp
app.register_blueprint(identity_bp)

from routes.browse import browse_bp  # co-browsing bridge (#154)
app.register_blueprint(browse_bp)

from routes.services import services_bp
app.register_blueprint(services_bp)

# Google Maps — /api/maps/config (JS API key for canvas pages) + /api/maps/directions
from routes.maps import maps_bp
app.register_blueprint(maps_bp)

# Camera Capture — flagship photo/album system (CompanyCam competitor)
from routes.capture import capture_bp
app.register_blueprint(capture_bp, url_prefix='/api/capture')

# WO-1.3 — wire the STT provider registry. autodiscover() loads
# config/providers.yaml and imports the STT provider modules so their
# registry.register() calls fire; the Service Catalog then reads STT ids from
# the SAME registry consumed by /api/stt/*. LLM entries in the yaml are
# DEPRECATED (see providers/llm/DEPRECATED.md) — importing them registers
# classes with zero call sites, harmless. Failure is non-fatal: the STT catalog
# falls back to the static handler-derived list.
try:
    from providers.registry import registry as _provider_registry
    _provider_registry.autodiscover()
    logger.info("Provider registry autodiscover complete (STT wired).")
except Exception as _e:
    logger.warning(f"Provider registry autodiscover failed (non-critical): {_e}")

from services.plugins import load_plugins
load_plugins(app)

# Auto-sync canvas manifest on startup so any pages written outside the API
# are picked up immediately without a restart.
try:
    sync_canvas_manifest()
    logger.info("Canvas manifest synced on startup.")
except Exception as _e:
    logger.warning(f"Canvas manifest auto-sync failed (non-critical): {_e}")

# Start canvas page version watcher (auto-saves versions when pages change)
try:
    from services.canvas_versioning import start_version_watcher
    start_version_watcher()
    logger.info("Canvas version watcher started.")
except Exception as _e:
    logger.warning(f"Canvas version watcher failed to start (non-critical): {_e}")


# ---------------------------------------------------------------------------
# Voice session management
# ---------------------------------------------------------------------------

from services.paths import VOICE_SESSION_FILE as _VSF_PATH, DB_PATH, UPLOADS_DIR
VOICE_SESSION_FILE = Path(_VSF_PATH)
_consecutive_empty_responses = 0


def _save_session_counter(counter: int) -> None:
    VOICE_SESSION_FILE.write_text(str(counter))


def get_voice_session_key() -> str:
    """Return the current voice session key, e.g. 'voice-main-6'."""
    prefix = os.getenv("VOICE_SESSION_PREFIX", "voice-main")
    try:
        counter = int(VOICE_SESSION_FILE.read_text().strip())
    except (FileNotFoundError, ValueError):
        counter = 1
        _save_session_counter(counter)
    return f"{prefix}-{counter}"


def bump_voice_session() -> str:
    """Increment the session counter and return the new session key."""
    global _consecutive_empty_responses
    prefix = os.getenv("VOICE_SESSION_PREFIX", "voice-main")
    try:
        counter = int(VOICE_SESSION_FILE.read_text().strip())
    except (FileNotFoundError, ValueError):
        counter = 1
    counter += 1
    _save_session_counter(counter)
    _consecutive_empty_responses = 0
    new_key = f"{prefix}-{counter}"
    logger.info(f"Session bumped → {new_key}")
    return new_key


# ---------------------------------------------------------------------------
# User usage tracking (SQLite)
# ---------------------------------------------------------------------------

MONTHLY_LIMIT = int(os.getenv("MONTHLY_USAGE_LIMIT", "20"))
UNLIMITED_USERS: list = [
    u.strip() for u in os.getenv("UNLIMITED_USER_IDS", "").split(",") if u.strip()
]
from services.db_pool import SQLitePool
db_pool = SQLitePool(DB_PATH, pool_size=5)


def init_db() -> None:
    conn = sqlite3.connect(DB_PATH)
    conn.execute("PRAGMA journal_mode=WAL")
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS usage (
            user_id TEXT PRIMARY KEY,
            message_count INTEGER DEFAULT 0,
            month TEXT,
            updated_at TEXT
        )
    """)
    c.execute("""
        CREATE TABLE IF NOT EXISTS conversation_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            session_id TEXT DEFAULT 'default',
            role TEXT NOT NULL,
            message TEXT NOT NULL,
            tts_provider TEXT,
            voice TEXT,
            created_at TEXT
        )
    """)
    c.execute("""
        CREATE TABLE IF NOT EXISTS conversation_metrics (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            session_id TEXT DEFAULT 'default',
            profile TEXT,
            model TEXT,
            handshake_ms INTEGER,
            llm_inference_ms INTEGER,
            tts_generation_ms INTEGER,
            total_ms INTEGER,
            user_message_len INTEGER,
            response_len INTEGER,
            tts_text_len INTEGER,
            tts_provider TEXT,
            tts_success INTEGER DEFAULT 1,
            tts_error TEXT,
            tool_count INTEGER DEFAULT 0,
            fallback_used INTEGER DEFAULT 0,
            error TEXT,
            created_at TEXT
        )
    """)
    conn.commit()
    conn.close()


def get_current_month() -> str:
    return datetime.now().strftime("%Y-%m")


def get_user_usage(user_id: str) -> int:
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("SELECT message_count, month FROM usage WHERE user_id = ?", (user_id,))
    row = c.fetchone()
    conn.close()
    if row:
        count, month = row
        return count if month == get_current_month() else 0
    return 0


def increment_usage(user_id: str) -> None:
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    current_month = get_current_month()
    now = datetime.now().isoformat()
    c.execute("SELECT month FROM usage WHERE user_id = ?", (user_id,))
    row = c.fetchone()
    if row:
        if row[0] != current_month:
            c.execute(
                "UPDATE usage SET message_count = 1, month = ?, updated_at = ? WHERE user_id = ?",
                (current_month, now, user_id),
            )
        else:
            c.execute(
                "UPDATE usage SET message_count = message_count + 1, updated_at = ? WHERE user_id = ?",
                (now, user_id),
            )
    else:
        c.execute(
            "INSERT INTO usage (user_id, message_count, month, updated_at) VALUES (?, 1, ?, ?)",
            (user_id, current_month, now),
        )
    conn.commit()
    conn.close()


init_db()


# ---------------------------------------------------------------------------
# Upload directory
# ---------------------------------------------------------------------------

UPLOADS_DIR.mkdir(parents=True, exist_ok=True)


# ---------------------------------------------------------------------------
# Routes — index
# ---------------------------------------------------------------------------

@app.route("/")
def serve_index():
    """Serve index.html with injected runtime config.

    Set AGENT_SERVER_URL in .env to override the backend URL the frontend
    connects to. Defaults to window.location.origin (correct for same-origin
    deployments).
    """
    import pathlib
    html = pathlib.Path("index.html").read_text()
    server_url = os.environ.get("AGENT_SERVER_URL", "").strip().rstrip("/")
    clerk_key = (os.environ.get("CLERK_PUBLISHABLE_KEY") or os.environ.get("VITE_CLERK_PUBLISHABLE_KEY", "")).strip()
    client_name = os.environ.get("CLIENT_NAME", "").strip()
    import json as _json
    devsite_map_raw = os.environ.get("DEVSITE_MAP", "{}").strip()
    try:
        devsite_map = _json.loads(devsite_map_raw)
    except Exception:
        devsite_map = {}
    config_parts = []
    config_parts.append(f'serverUrl:"{server_url}"' if server_url else 'serverUrl:window.location.origin')
    if clerk_key:
        config_parts.append(f'clerkPublishableKey:"{clerk_key}"')
    if devsite_map:
        config_parts.append(f'devsiteMap:{_json.dumps(devsite_map)}')
    if client_name:
        config_parts.append(f'clientName:{_json.dumps(client_name)}')
        config_parts.append('managedUpdates:true')
    config_block = f'<script>window.AGENT_CONFIG={{{",".join(config_parts)}}};</script>'
    html = html.replace("<head>", f"<head>\n  {config_block}", 1)
    # Replace PWA title and apple-mobile-web-app-title with client name
    if client_name:
        html = html.replace("<title>OpenVoiceUI</title>", f"<title>{client_name}</title>")
        html = html.replace('content="OpenVoiceUI"', f'content="{client_name}"')
    resp = Response(html, mimetype="text/html")
    resp.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    return resp


# ---------------------------------------------------------------------------
# Routes — version info
# ---------------------------------------------------------------------------

_VERSION_INFO = {"commit": "unknown", "branch": "unknown", "date": "unknown"}
_version_file = Path(__file__).parent / "version.json"
if _version_file.exists():
    try:
        _VERSION_INFO = json.loads(_version_file.read_text())
    except Exception:
        pass

# Read package.json version as authoritative version string
_PACKAGE_VERSION = "unknown"
_package_file = Path(__file__).parent / "package.json"
if _package_file.exists():
    try:
        _PACKAGE_VERSION = json.loads(_package_file.read_text()).get("version", "unknown")
    except Exception:
        pass
_VERSION_INFO["version"] = _PACKAGE_VERSION


@app.route("/api/config", methods=["GET"])
def get_public_config():
    """Public client config — only values safe to expose pre-auth.
    admin.html bootstraps Clerk from this (it is served statically and does not
    get the window.AGENT_CONFIG injection index.html gets)."""
    clerk_key = (os.getenv("CLERK_PUBLISHABLE_KEY") or os.getenv("VITE_CLERK_PUBLISHABLE_KEY", "")).strip()
    return jsonify({"clerkPublishableKey": clerk_key})


@app.route("/api/version", methods=["GET"])
def get_version():
    """Return build version and check GitHub for latest release."""
    data = {**_VERSION_INFO, "uptime_seconds": round(time.time() - SERVER_START_TIME)}
    # Check for newer release on GitHub (cached, non-blocking)
    latest = _get_latest_release_info()
    if latest:
        data["latest_commit"] = latest["sha"]
        data["latest_date"] = latest["date"]
        data["latest_message"] = latest["message"]
        data["latest_version"] = latest.get("latest_version", "")
        # Check if we need an update. Three strategies:
        # 1. git merge-base (if .git exists) — handles being ahead of release
        # 2. Commit SHA comparison — direct match
        # 3. package.json version vs release tag — fallback when commit unknown
        current = _VERSION_INFO.get("commit", "unknown")
        data["update_available"] = False
        if current != "unknown" and latest.get("sha"):
            app_dir = Path(__file__).parent
            if (app_dir / ".git").is_dir():
                try:
                    result = subprocess.run(
                        ["git", "merge-base", "--is-ancestor",
                         latest["sha"], "HEAD"],
                        cwd=str(app_dir), capture_output=True, timeout=5,
                    )
                    # returncode 0 = release is ancestor of HEAD (up to date)
                    # returncode 1 = release is NOT ancestor (need update)
                    data["update_available"] = result.returncode != 0
                except Exception:
                    data["update_available"] = not latest["sha"].startswith(current)
            else:
                data["update_available"] = not latest["sha"].startswith(current)
        elif current == "unknown" and latest.get("latest_version"):
            # Commit unknown (Docker build without args, etc.)
            # Fall back to comparing package.json version against release tag.
            pkg_ver = _PACKAGE_VERSION
            rel_tag = latest["latest_version"]
            if pkg_ver != "unknown" and rel_tag:
                rel_ver = rel_tag.lstrip("v")
                data["update_available"] = rel_ver.split("-")[0] != pkg_ver.split("-")[0] or rel_ver > pkg_ver
    return jsonify(data)


# Cache GitHub check for 5 minutes to avoid hammering the API
_github_cache = {"data": None, "expires": 0}


_GITHUB_REPO = os.environ.get("GITHUB_REPO", "MCERQUA/OpenVoiceUI").strip()


def _get_latest_release_info():
    """Fetch the latest GitHub release and its commit SHA.

    Only compares against tagged releases — not every commit on main.
    This prevents false "update available" banners from routine merges.
    """
    now = time.time()
    if _github_cache["data"] and now < _github_cache["expires"]:
        return _github_cache["data"]
    try:
        # Get the latest release
        rel = requests.get(
            f"https://api.github.com/repos/{_GITHUB_REPO}/releases/latest",
            headers={"Accept": "application/vnd.github.v3+json"},
            timeout=5,
        )
        if rel.status_code != 200:
            return _github_cache.get("data")

        rel_data = rel.json()
        tag_name = rel_data.get("tag_name", "")

        # Get the commit SHA for this release tag
        tag_resp = requests.get(
            f"https://api.github.com/repos/{_GITHUB_REPO}/git/ref/tags/{tag_name}",
            headers={"Accept": "application/vnd.github.v3+json"},
            timeout=5,
        )
        sha = ""
        if tag_resp.status_code == 200:
            tag_obj = tag_resp.json().get("object", {})
            sha = tag_obj.get("sha", "")[:7]

        result = {
            "sha": sha,
            "date": rel_data.get("published_at", ""),
            "message": rel_data.get("name", tag_name),
            "latest_version": tag_name,
        }
        _github_cache["data"] = result
        _github_cache["expires"] = now + 300  # 5 min cache
        return result
    except Exception:
        pass
    return _github_cache.get("data")


def _self_update():
    """Intelligent update: analyse → detect agent → review → apply → verify.

    Instead of a blind ``git pull``, this:
    1. Analyses what changed upstream vs what's customised locally
    2. Searches for an available CLI coding agent (claude, codex, z-code, etc.)
    3. If an agent is found AND there are conflicts → spawns it with a
       comprehensive prompt that enumerates every possible breaking point
    4. If no agent or low risk → uses heuristic-based smart update with
       automatic backup/rollback
    5. Verifies health after every update; rolls back on failure

    Works in all deployment scenarios: native, Docker, multi-tenant, Pinokio.
    """
    from services.update_manager import UpdateManager

    app_dir = Path(__file__).parent
    mgr = UpdateManager(app_dir)
    return mgr.apply_update()


@app.route("/api/version/preview", methods=["GET"])
def preview_update():
    """Return a preview of what an update would change, without applying it.

    The frontend can call this to show the user:
    - What files will change
    - What local customisations exist
    - What conflicts were detected
    - Which update method will be used (AI agent or smart fallback)
    - Risk level (low / medium / high)
    """
    from services.update_manager import UpdateManager

    app_dir = Path(__file__).parent
    mgr = UpdateManager(app_dir)
    return jsonify(mgr.get_update_preview())


@app.route("/api/version/update", methods=["POST"])
def trigger_update():
    """Update the app to the latest version.

    Tries three strategies in order:
    1. Intelligent self-update (analyses diffs, detects agent, backs up,
       verifies) — works for native/Pinokio/dev/Docker installs with .git
    2. Host update service (JamBot / managed Docker hosting)
    3. Returns instructions as last resort
    """
    # Check if already current
    latest = _get_latest_release_info()
    current = _VERSION_INFO.get("commit", "unknown")
    if latest and current != "unknown" and latest["sha"].startswith(current):
        return jsonify({"status": "current", "message": "Already up to date"})

    # Strategy 1: Intelligent self-update if this is a git repo
    app_dir = Path(__file__).parent
    if (app_dir / ".git").is_dir():
        result = _self_update()
        status = result.get("status", "error")

        if status == "success":
            return jsonify({
                "status": "updating",
                "message": "Update applied — app will restart in a moment",
                "method": result.get("method", "unknown"),
                "details": {
                    "files_updated": len(result.get("files_updated", [])),
                    "customisations_preserved": result.get("customisations_preserved", []),
                    "warnings": result.get("warnings", []),
                },
            })
        elif status == "current":
            return jsonify({"status": "current", "message": "Already up to date"})
        elif status == "rolled_back":
            return jsonify({
                "status": "rolled_back",
                "message": "Update was rolled back due to issues",
                "reason": result.get("reason", "Verification failed"),
                "method": result.get("method", "unknown"),
            }), 409
        else:
            return jsonify({
                "status": "error",
                "error": result.get("reason", result.get("error", "Update failed")),
                "method": result.get("method", "unknown"),
            }), 500

    # Strategy 2: Host update service (JamBot / managed Docker)
    client_name = os.environ.get("CLIENT_NAME", "").strip()
    if client_name:
        for host in ["host.docker.internal", "172.17.0.1"]:
            try:
                r = requests.post(f"http://{host}:5199/update/{client_name}", timeout=5)
                if r.status_code == 200:
                    return jsonify({"status": "updating", "message": "Updating — your system will restart in a moment"})
            except Exception:
                pass

    # Strategy 3: Manual — shouldn't happen often
    return jsonify({"status": "manual", "message": "Pull the latest version and rebuild to update"})


# ---------------------------------------------------------------------------
# Routes — health probes
# ---------------------------------------------------------------------------

from services.health import health_checker as _health_checker


@app.route("/health/live", methods=["GET"])
def health_live():
    """Liveness probe — always 200 while the process is running."""
    result = _health_checker.liveness()
    return jsonify({"healthy": result.healthy, "message": result.message, "details": result.details}), 200


@app.route("/health/ready", methods=["GET"])
def health_ready():
    """Readiness probe — 200 only when Gateway and TTS are available."""
    result = _health_checker.readiness()
    code = 200 if result.healthy else 503
    return jsonify({"healthy": result.healthy, "message": result.message, "details": result.details}), code


@app.route("/api/memory-status", methods=["GET"])
def memory_status():
    """Process memory usage — for watchdog monitoring."""
    import resource
    rusage = resource.getrusage(resource.RUSAGE_SELF)
    current_mb = rusage.ru_maxrss / 1024  # ru_maxrss is KB on Linux
    return jsonify({"process": {"current_mb": round(current_mb, 1)}})


# ---------------------------------------------------------------------------
# Routes — session
# ---------------------------------------------------------------------------

@app.route("/api/session", methods=["GET"])
def session_info():
    """Return the current voice session key and consecutive-empty-response count."""
    return jsonify({
        "sessionKey": get_voice_session_key(),
        "consecutiveEmpty": _consecutive_empty_responses,
    })


@app.route("/api/session/reset", methods=["POST"])
def session_reset():
    """Reset the voice session context.

    Body (JSON, optional):
      { "mode": "soft" }  — bump session key only (default)
      { "mode": "hard" }  — bump session key and pre-warm the new session
    """
    from services.gateway import gateway_connection

    data = request.get_json(silent=True) or {}
    mode = data.get("mode", "soft")
    if mode not in ("soft", "hard"):
        return jsonify({"error": f"Invalid mode '{mode}'. Use 'soft' or 'hard'."}), 400

    old_key = get_voice_session_key()
    new_key = bump_voice_session()

    if mode == "hard":
        def _prewarm():
            try:
                gateway_connection.stream_to_queue(
                    queue.Queue(),
                    "[SYSTEM: session pre-warm, reply with exactly: ok]",
                    new_key,
                    [],
                )
                logger.info(f"Pre-warm complete for {new_key}")
            except Exception as e:
                logger.warning(f"Pre-warm failed: {e}")
        threading.Thread(target=_prewarm, daemon=True).start()

    return jsonify({
        "old": old_key,
        "new": new_key,
        "mode": mode,
        "message": f"Session reset ({mode})." + (" Pre-warming new session..." if mode == "hard" else ""),
    })


# ---------------------------------------------------------------------------
# Routes — diagnostics
# ---------------------------------------------------------------------------

@app.route("/api/diagnostics", methods=["GET"])
def diagnostics():
    """Diagnostic dashboard — uptime, active config, recent timing metrics."""
    import resource

    uptime_seconds = int(time.time() - SERVER_START_TIME)
    uptime_h = uptime_seconds // 3600
    uptime_m = (uptime_seconds % 3600) // 60
    rusage = resource.getrusage(resource.RUSAGE_SELF)
    memory_mb = round(rusage.ru_maxrss / 1024, 1)

    state = {
        "server": {
            "uptime": f"{uptime_h}h {uptime_m}m",
            "uptime_seconds": uptime_seconds,
            "memory_mb": memory_mb,
            "pid": os.getpid(),
            "started_at": datetime.fromtimestamp(SERVER_START_TIME).isoformat(),
        },
        "config": {
            "gateway_url": os.getenv("CLAWDBOT_GATEWAY_URL", "ws://127.0.0.1:18791"),
            "session_key": get_voice_session_key(),
            "tts_provider": os.getenv("DEFAULT_TTS_PROVIDER", "groq"),
            "port": os.getenv("PORT", "5001"),
        },
    }

    try:
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row
        c = conn.cursor()
        c.execute("""
            SELECT profile, model, handshake_ms, llm_inference_ms,
                   tts_generation_ms, total_ms, user_message_len,
                   response_len, tts_text_len, tts_provider, tts_success,
                   tts_error, tool_count, fallback_used, error, created_at
            FROM conversation_metrics
            ORDER BY id DESC LIMIT 10
        """)
        state["recent_conversations"] = [dict(r) for r in c.fetchall()]
        c.execute("""
            SELECT COUNT(*) as total_conversations,
                   AVG(total_ms) as avg_total_ms,
                   AVG(llm_inference_ms) as avg_llm_ms,
                   AVG(tts_generation_ms) as avg_tts_ms,
                   AVG(handshake_ms) as avg_handshake_ms,
                   SUM(CASE WHEN tts_success = 0 THEN 1 ELSE 0 END) as tts_failures,
                   SUM(CASE WHEN error IS NOT NULL THEN 1 ELSE 0 END) as errors,
                   MAX(total_ms) as max_total_ms,
                   MIN(total_ms) as min_total_ms
            FROM conversation_metrics
            WHERE created_at > datetime('now', '-1 hour')
        """)
        stats = dict(c.fetchone() or {})
        for key in ("avg_total_ms", "avg_llm_ms", "avg_tts_ms", "avg_handshake_ms"):
            if stats.get(key) is not None:
                stats[key] = round(stats[key])
        state["last_hour_stats"] = stats
        conn.close()
    except Exception as e:
        state["metrics_error"] = str(e)

    return jsonify(state)


# ---------------------------------------------------------------------------
# Routes — Hume EVI token (used by src/adapters/hume-evi.js)
# ---------------------------------------------------------------------------

@app.route("/api/hume/token", methods=["GET"])
def get_hume_token():
    """Return a short-lived Hume access token for EVI WebSocket connections.

    Returns 403 when Hume credentials are not configured — the frontend
    adapter treats this as 'Hume unavailable' rather than an error.
    """
    api_key = os.getenv("HUME_API_KEY")
    secret_key = os.getenv("HUME_SECRET_KEY")

    if not api_key or not secret_key:
        return jsonify({"error": "Hume API credentials not configured", "available": False}), 403

    try:
        credentials = f"{api_key}:{secret_key}"
        encoded = base64.b64encode(credentials.encode()).decode()
        response = requests.post(
            "https://api.hume.ai/oauth2-cc/token",
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "Authorization": f"Basic {encoded}",
            },
            data={"grant_type": "client_credentials"},
            timeout=10,
        )
        if response.status_code != 200:
            logger.error(f"Hume token request failed: {response.status_code} — {response.text}")
            return jsonify({"error": "Failed to get Hume access token", "available": False}), 500
        token_data = response.json()
        return jsonify({
            "access_token": token_data.get("access_token"),
            "expires_in": token_data.get("expires_in", 3600),
            "config_id": os.getenv("HUME_CONFIG_ID"),
            "available": True,
        })
    except Exception as e:
        logger.error(f"Hume token error: {e}")
        return jsonify({"error": "Failed to retrieve token", "available": False}), 500


# ---------------------------------------------------------------------------
# Routes — xAI Grok Realtime config (used by src/adapters/xai-realtime.js)
# ---------------------------------------------------------------------------

@app.route("/api/xai/config", methods=["GET"])
def get_xai_config():
    """Check whether xAI Realtime is configured server-side.

    Returns {"available": true} when XAI_API_KEY is set.
    The frontend adapter calls this during init so it can surface a helpful
    message instead of silently failing on WebSocket connect.
    """
    api_key = os.getenv("XAI_API_KEY")
    return jsonify({"available": bool(api_key)})


# ---------------------------------------------------------------------------
# Routes — STT (Speech-to-Text)
# ---------------------------------------------------------------------------

@app.route("/api/stt/groq", methods=["POST"])
def groq_stt():
    """Transcribe audio using Groq Whisper Large v3 Turbo (cloud, fast)."""
    from services.tts import get_groq_client as _get_groq_client

    if "audio" not in request.files:
        return jsonify({"error": "No audio file provided"}), 400

    audio_file = request.files["audio"]
    groq = _get_groq_client()
    if not groq:
        return jsonify({"error": "Groq client not available — check GROQ_API_KEY"}), 500

    try:
        audio_bytes = audio_file.read()
        audio_tuple = (
            audio_file.filename or "audio.webm",
            audio_bytes,
            audio_file.content_type or "audio/webm",
        )
        transcription = groq.audio.transcriptions.create(
            file=audio_tuple,
            model="whisper-large-v3-turbo",
            response_format="verbose_json",
            language="en",
            temperature=0,
            prompt="",
        )
        # Filter segments with high no_speech_prob or low confidence (Whisper hallucinations)
        import re as _re
        segments = getattr(transcription, 'segments', None)
        if segments:
            filtered_texts = []
            for seg in segments:
                _nsp = seg.get('no_speech_prob', 0) if isinstance(seg, dict) else getattr(seg, 'no_speech_prob', 0)
                _alp = seg.get('avg_logprob', 0) if isinstance(seg, dict) else getattr(seg, 'avg_logprob', 0)
                _stxt = (seg.get('text', '') if isinstance(seg, dict) else seg.text).strip()
                # Reject: high no-speech probability OR very low confidence
                if _nsp >= 0.2:
                    logger.debug(f"Groq STT: dropping segment (no_speech_prob={_nsp:.2f}): {_stxt!r}")
                    continue
                if _alp < -1.0:
                    logger.debug(f"Groq STT: dropping segment (avg_logprob={_alp:.2f}): {_stxt!r}")
                    continue
                filtered_texts.append(_stxt)
            text = ' '.join(filtered_texts).strip()
        else:
            text = (transcription.text or "").strip()
        logger.info(f"Groq STT: {text!r}")

        # --- Whisper hallucination filtering ---
        _WHISPER_HALLUCINATIONS = {
            "thank you", "thanks for watching", "thanks for listening",
            "i'm here with closed captioning", "closed captioning",
            "subscribe", "please subscribe", "like and subscribe",
            "you", "bye", "the end", "subtitles by", "translated by",
            "voice command for ai assistant", "voice command for ai",
            "alright", "all right", "okay", "ok", "yeah", "yes",
            "um", "uh", "hmm", "huh", "oh", "ah",
            "so", "well", "right", "sure", "hey",
            "thanks", "thank you so much",
            "i don't know", "i'm sorry",
        }
        # Substrings that indicate prompt-echo or known garbage
        _HALLUCINATION_SUBSTRINGS = [
            "voice command for ai",
            "thanks for watching", "thanks for listening",
            "like and subscribe", "please subscribe",
            "subtitles by", "translated by", "closed captioning",
            "coupo foundation",  # known recurring hallucination
        ]
        text_lower = text.lower().rstrip('.!?,;:')
        _meaningful = _re.sub(r'[^a-zA-Z0-9]', '', text)

        def _is_hallucination(t, t_lower):
            # Exact match against known phrases
            if t_lower in _WHISPER_HALLUCINATIONS:
                return True
            # Too short to be real speech
            if len(_meaningful) < 3:
                return True
            # Prompt text or known garbage appears anywhere in transcription
            for sub in _HALLUCINATION_SUBSTRINGS:
                if sub in t_lower:
                    return True
            # Repetitive pattern: same word/phrase repeated many times
            words = _re.findall(r'[a-zA-Z]+', t)
            if len(words) >= 4:
                from collections import Counter
                counts = Counter(w.lower() for w in words)
                most_common_count = counts.most_common(1)[0][1]
                if most_common_count / len(words) >= 0.5:
                    return True
            return False

        if _is_hallucination(text, text_lower):
            logger.info(f"Groq STT: FILTERED hallucination/garbage: {text!r}")
            return jsonify({"transcript": "", "success": True, "filtered": True})

        return jsonify({"transcript": text, "success": True})
    except Exception as e:
        logger.error(f"Groq STT error: {e}")
        return jsonify({"error": "Speech-to-text failed"}), 500


@app.route("/api/stt/deepgram/token", methods=["GET"])
def deepgram_stt_token():
    """Return the Deepgram API key for browser-side WebSocket streaming.

    The browser needs the key to open a direct WebSocket to Deepgram's
    live transcription API.  The key is passed via the WebSocket sub-protocol
    header so it never appears in URLs or logs.

    TODO(scoped-key): Deepgram supports scoped / short-lived project keys via
    its /v1/projects/{id}/keys grant API. We should mint a temporary key with
    only 'usage:write' scope + a short TTL per session instead of handing out
    the long-lived project key. Until that helper exists we return the
    configured key ONLY to authenticated callers (Clerk session or agent key).

    SECURITY: the raw key must never be returned unauthenticated. The public
    allowlist no longer exposes /api/stt/* (require_auth gates it), and this
    in-handler check is defense-in-depth so the key can't leak even if the
    gate is ever loosened.
    """
    # Defense-in-depth auth check (only enforced when Clerk auth is configured;
    # local/self-hosted mode with no Clerk key runs fully open per app.py).
    _clerk_configured = bool(
        (os.getenv("CLERK_PUBLISHABLE_KEY") or os.getenv("VITE_CLERK_PUBLISHABLE_KEY", "")).strip()
    )
    if _clerk_configured:
        _agent_key = os.environ.get("AGENT_API_KEY", "").strip()
        _is_agent = bool(_agent_key) and request.headers.get("X-Agent-Key") == _agent_key
        if not _is_agent:
            from services.auth import get_token_from_request, verify_clerk_token
            _token = get_token_from_request()
            _user = verify_clerk_token(_token) if _token else None
            if not _user:
                return jsonify({"error": "Unauthorized", "code": "auth_required"}), 401

    api_key = os.environ.get("DEEPGRAM_API_KEY", "")
    if not api_key:
        return jsonify({"error": "DEEPGRAM_API_KEY not configured"}), 500
    return jsonify({"token": api_key})


@app.route("/api/stt/deepgram", methods=["POST"])
def deepgram_stt():
    """Transcribe audio using Deepgram Nova-2 API (reliable, low-cost)."""
    import re as _re

    if "audio" not in request.files:
        return jsonify({"error": "No audio file provided"}), 400

    api_key = os.environ.get("DEEPGRAM_API_KEY", "")
    if not api_key:
        return jsonify({"error": "DEEPGRAM_API_KEY not configured"}), 500

    audio_file = request.files["audio"]
    try:
        audio_bytes = audio_file.read()
        content_type = audio_file.content_type or "audio/webm"

        import requests as _requests
        resp = _requests.post(
            "https://api.deepgram.com/v1/listen",
            params={
                "model": "nova-2",
                "language": "en",
                "smart_format": "true",
                "punctuate": "true",
            },
            headers={
                "Authorization": f"Token {api_key}",
                "Content-Type": content_type,
            },
            data=audio_bytes,
            timeout=15,
        )

        # JamBot Books: Deepgram uses `requests` (no httpx SDK to attach), so
        # record the STT call explicitly (file-drop leg). Fire-and-forget.
        try:
            from services.jambot_books_hook import record_provider_call
            record_provider_call('deepgram', endpoint='/v1/listen', op='stt',
                                 units=str(len(audio_bytes)), status=resp.status_code,
                                 model='nova-2')
        except Exception:
            pass

        if resp.status_code != 200:
            logger.error(f"Deepgram API error {resp.status_code}: {resp.text[:300]}")
            return jsonify({"error": f"Deepgram API error: {resp.status_code}"}), 502

        result = resp.json()
        channels = result.get("results", {}).get("channels", [])
        if not channels:
            return jsonify({"transcript": "", "success": True})

        alt = channels[0].get("alternatives", [{}])[0]
        text = alt.get("transcript", "").strip()
        confidence = alt.get("confidence", 0)

        logger.info(f"Deepgram STT: {text!r} (confidence={confidence:.2f})")

        # Low confidence filter
        if confidence < 0.3 and text:
            logger.info(f"Deepgram STT: FILTERED low confidence ({confidence:.2f}): {text!r}")
            return jsonify({"transcript": "", "success": True, "filtered": True})

        # Hallucination filtering (same as Groq)
        _HALLUCINATIONS = {
            "thank you", "thanks for watching", "thanks for listening",
            "subscribe", "please subscribe", "like and subscribe",
            "the end", "subtitles by", "translated by", "closed captioning",
            "voice command for ai assistant", "voice command for ai",
            "thanks", "thank you so much",
        }
        _HALLUCINATION_SUBSTRINGS = [
            "voice command for ai", "thanks for watching", "thanks for listening",
            "like and subscribe", "please subscribe",
            "subtitles by", "translated by", "closed captioning",
        ]
        text_lower = text.lower().rstrip('.!?,;:')
        meaningful = _re.sub(r'[^a-zA-Z0-9]', '', text)

        if text_lower in _HALLUCINATIONS:
            logger.info(f"Deepgram STT: FILTERED hallucination: {text!r}")
            return jsonify({"transcript": "", "success": True, "filtered": True})
        if len(meaningful) < 3:
            logger.info(f"Deepgram STT: FILTERED too short: {text!r}")
            return jsonify({"transcript": "", "success": True, "filtered": True})
        for sub in _HALLUCINATION_SUBSTRINGS:
            if sub in text_lower:
                logger.info(f"Deepgram STT: FILTERED hallucination substring: {text!r}")
                return jsonify({"transcript": "", "success": True, "filtered": True})

        return jsonify({"transcript": text, "success": True, "confidence": confidence})
    except Exception as e:
        logger.error(f"Deepgram STT error: {e}")
        return jsonify({"error": "Speech-to-text failed"}), 500


@app.route("/api/stt/local", methods=["POST"])
def local_stt():
    """Transcribe audio using local Faster-Whisper with Silero VAD.

    Requires faster-whisper and ffmpeg. Uses the 'tiny' model to keep
    memory usage low.
    """
    if "audio" not in request.files:
        return jsonify({"error": "No audio file provided"}), 400

    audio_file = request.files["audio"]
    audio_bytes = audio_file.read()

    # PREFERRED PATH: the shared jambot-parakeet service. One ~680MB model for
    # the whole fleet instead of a copy per OVU container, and it does its own
    # ffmpeg normalisation. Falls through to the in-process path below only if
    # the service is unreachable — a missing shared service degrades, it does
    # not break voice.
    _text = _parakeet_transcribe(audio_bytes, audio_file.filename or "audio.webm")
    if _text is not None:
        logger.info("Local STT (parakeet): %r", _text)
        return jsonify({"transcript": _text, "success": True, "engine": "parakeet-tdt-0.6b-v3"})

    with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as tmp:
        tmp.write(audio_bytes)
        tmp_path = tmp.name

    try:
        wav_path = tmp_path.replace(".webm", ".wav")
        result = subprocess.run(
            ["ffmpeg", "-y", "-i", tmp_path, "-ar", "16000", "-ac", "1", "-f", "wav", wav_path],
            capture_output=True,
            timeout=10,
        )
        if result.returncode != 0:
            logger.warning(f"FFmpeg conversion failed, transcribing original: {result.stderr.decode()}")
            wav_path = tmp_path

        segments, info = get_whisper_model().transcribe(
            wav_path,
            language="en",
            vad_filter=True,
            vad_parameters={"min_silence_duration_ms": 500, "threshold": 0.5},
        )
        transcript = " ".join(seg.text for seg in segments).strip()
        logger.info(f"Local STT: {transcript!r} ({info.duration:.1f}s)")
        return jsonify({"transcript": transcript, "success": True})
    except Exception as e:
        logger.error(f"Local STT error: {e}")
        return jsonify({"error": "Speech-to-text failed"}), 500
    finally:
        for f in [tmp_path, tmp_path.replace(".webm", ".wav")]:
            try:
                os.unlink(f)
            except OSError:
                pass


@app.route("/api/stt/external", methods=["POST"])
def external_stt():
    """Transcribe audio using an external STT service (bring your own).

    Sends audio to the URL configured in STT_API_URL. Supports OpenAI-compatible
    and generic Whisper ASR formats. Configure via .env or admin panel.
    """
    if "audio" not in request.files:
        return jsonify({"error": "No audio file provided"}), 400

    audio_file = request.files["audio"]
    audio_bytes = audio_file.read()
    language = request.form.get("language", "en")

    try:
        from providers.stt.external_provider import ExternalSTTProvider
        provider = ExternalSTTProvider()
        result = provider.transcribe(audio_bytes, language=language)

        if not result.text:
            return jsonify({"transcript": "", "success": True, "filtered": True})

        return jsonify({
            "transcript": result.text,
            "success": True,
            "confidence": result.confidence,
            "duration_ms": result.duration_ms,
        })
    except Exception as e:
        logger.error(f"External STT error: {e}")
        return jsonify({"error": f"External STT failed: {e}"}), 500


# ---------------------------------------------------------------------------
# Routes — web search
# ---------------------------------------------------------------------------

@app.route("/api/search/brave", methods=["GET"])
def brave_search():
    """Web search via Brave Search API. Requires BRAVE_API_KEY in .env."""
    query = request.args.get("q", "").strip()
    if not query:
        return jsonify({"error": "No query provided"}), 400

    brave_api_key = os.getenv("BRAVE_API_KEY")
    if not brave_api_key:
        return jsonify({"error": "BRAVE_API_KEY not configured"}), 500

    try:
        response = requests.get(
            "https://api.search.brave.com/res/v1/web/search",
            headers={"Accept": "application/json", "X-Subscription-Token": brave_api_key},
            params={"q": query, "count": 10, "search_lang": "en", "freshness": "pw"},
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()
        results = [
            {
                "title": r.get("title", ""),
                "url": r.get("url", ""),
                "description": r.get("description", ""),
            }
            for r in data.get("web", {}).get("results", [])[:5]
        ]
        return jsonify({"query": query, "results": results, "success": True})
    except Exception as e:
        logger.error(f"Brave Search error: {e}")
        return jsonify({"error": "Search failed"}), 500


@app.route("/api/search", methods=["GET", "POST"])
def web_search():
    """Web search via DuckDuckGo routed through residential SOCKS5 proxy."""
    import urllib.parse
    from html.parser import HTMLParser

    if request.method == "POST":
        query = (request.get_json() or {}).get("query")
    else:
        query = request.args.get("query")

    if not query:
        return jsonify({"error": "query required"}), 400

    class _DDGParser(HTMLParser):
        def __init__(self):
            super().__init__()
            self.results = []
            self._current = {}
            self._capture = False
            self._text = ""

        def handle_starttag(self, tag, attrs):
            d = dict(attrs)
            if tag == "a" and d.get("class") == "result__a":
                self._current = {"url": d.get("href", ""), "title": "", "snippet": ""}
                self._capture = True
            elif tag == "a" and d.get("class") == "result__snippet":
                self._capture = True

        def handle_endtag(self, tag):
            if tag == "a" and self._capture:
                if self._current and not self._current.get("title"):
                    self._current["title"] = self._text.strip()
                elif self._current.get("title") and not self._current.get("snippet"):
                    self._current["snippet"] = self._text.strip()
                    if self._current["title"] and self._current["url"]:
                        self.results.append(self._current)
                    self._current = {}
                self._capture = False
                self._text = ""

        def handle_data(self, data):
            if self._capture:
                self._text += data

    # Route through residential proxy (avoids datacenter IP blocks)
    proxy_server = os.getenv("BROWSE_PROXY_SERVER", "")
    session_proxies = {"http": proxy_server, "https": proxy_server} if proxy_server else None

    try:
        encoded = urllib.parse.quote_plus(query)
        resp = requests.get(
            f"https://html.duckduckgo.com/html/?q={encoded}",
            headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"},
            proxies=session_proxies,
            timeout=15,
        )
        resp.raise_for_status()
        html = resp.text
        parser = _DDGParser()
        parser.feed(html)
        results = parser.results[:5]
        return jsonify({"query": query, "results": results, "success": True})
    except Exception as e:
        logger.error(f"DuckDuckGo search error: {e}")
        return jsonify({"error": "Search failed"}), 500


# ---------------------------------------------------------------------------
# Routes — usage quotas
# ---------------------------------------------------------------------------

@app.route("/api/usage/<user_id>", methods=["GET"])
def check_usage(user_id):
    """Return the current month's usage for a user."""
    # Clerk-authenticated users may only query their own usage.
    # Internal agent calls (X-Agent-Key, no g.clerk_user_id) are trusted.
    authenticated_user = g.get('clerk_user_id')
    if authenticated_user and authenticated_user != user_id:
        return jsonify({'error': 'Unauthorized', 'code': 'user_mismatch'}), 403
    if user_id in UNLIMITED_USERS:
        return jsonify({
            "user_id": user_id,
            "used": get_user_usage(user_id),
            "limit": -1,
            "remaining": -1,
            "allowed": True,
            "unlimited": True,
        })
    count = get_user_usage(user_id)
    return jsonify({
        "user_id": user_id,
        "used": count,
        "limit": MONTHLY_LIMIT,
        "remaining": max(0, MONTHLY_LIMIT - count),
        "allowed": count < MONTHLY_LIMIT,
    })


@app.route("/api/usage/<user_id>/increment", methods=["POST"])
def track_usage(user_id):
    """Increment usage count for a user (called after each agent response)."""
    authenticated_user = g.get('clerk_user_id')
    if authenticated_user and authenticated_user != user_id:
        return jsonify({'error': 'Unauthorized', 'code': 'user_mismatch'}), 403
    if user_id in UNLIMITED_USERS:
        increment_usage(user_id)
        return jsonify({
            "user_id": user_id,
            "used": get_user_usage(user_id),
            "limit": -1,
            "remaining": -1,
            "unlimited": True,
        })
    count = get_user_usage(user_id)
    if count >= MONTHLY_LIMIT:
        return jsonify({"error": "Monthly limit reached", "used": count, "limit": MONTHLY_LIMIT}), 429
    increment_usage(user_id)
    new_count = count + 1
    return jsonify({
        "user_id": user_id,
        "used": new_count,
        "limit": MONTHLY_LIMIT,
        "remaining": max(0, MONTHLY_LIMIT - new_count),
    })


# ---------------------------------------------------------------------------
# Routes — server commands (whitelisted)
# ---------------------------------------------------------------------------

ALLOWED_COMMANDS = {
    "git_status":     {"cmd": ["git", "status"],                                "desc": "Git working tree status"},
    "git_log":        {"cmd": ["git", "log", "--oneline", "-10"],               "desc": "Last 10 commits"},
    "disk_usage":     {"cmd": ["df", "-h", "/"],                                "desc": "Disk usage"},
    "memory":         {"cmd": ["free", "-h"],                                   "desc": "Memory usage"},
    "uptime":         {"cmd": ["uptime"],                                       "desc": "System uptime"},
    "date":           {"cmd": ["date"],                                         "desc": "Current date/time"},
    "whoami":         {"cmd": ["whoami"],                                       "desc": "Current user"},
    "nginx_status":   {"cmd": ["systemctl", "status", "nginx", "--no-pager"],   "desc": "Nginx status"},
    "service_status": {"cmd": ["systemctl", "status", "openvoiceui", "--no-pager"], "desc": "OpenVoiceUI service status"},
    "network":        {"cmd": ["ss", "-tuln"],                                  "desc": "Active network listeners"},
    "processes":      {"cmd": ["ps", "aux", "--sort=-%cpu"],                    "desc": "Running processes by CPU"},
    "hostname":       {"cmd": ["hostname"],                                     "desc": "Server hostname"},
    "ip_address":     {"cmd": ["hostname", "-I"],                               "desc": "Server IP addresses"},
}

_COMMAND_KEYWORDS = {
    "git": "git_status", "commit": "git_log", "disk": "disk_usage",
    "space": "disk_usage", "memory": "memory", "ram": "memory",
    "time": "date", "date": "date", "nginx": "nginx_status",
    "web": "nginx_status", "service": "service_status",
    "openvoiceui": "service_status", "network": "network",
    "ports": "network", "process": "processes", "cpu": "processes",
    "running": "processes", "host": "hostname", "ip": "ip_address",
    "address": "ip_address", "uptime": "uptime",
}


@app.route("/api/command", methods=["GET", "POST"])
def run_command():
    """Run a whitelisted server command. Accepts a command key or natural language."""
    if request.method == "POST":
        command = (request.get_json() or {}).get("command")
    else:
        command = request.args.get("command")

    if not command:
        return jsonify({
            "available_commands": [{"name": k, "description": v["desc"]} for k, v in ALLOWED_COMMANDS.items()],
        })

    key = command.lower().replace(" ", "_").replace("-", "_")
    matched = key if key in ALLOWED_COMMANDS else next(
        (v for k, v in _COMMAND_KEYWORDS.items() if k in key), None
    )

    if not matched:
        return jsonify({"error": "command not in whitelist", "available": list(ALLOWED_COMMANDS.keys())}), 400

    cmd_info = ALLOWED_COMMANDS[matched]
    try:
        result = subprocess.run(
            cmd_info["cmd"],
            capture_output=True,
            text=True,
            timeout=30,
            cwd=str(Path(__file__).parent),
        )
        output = (result.stdout.strip() or result.stderr.strip())[:1500]
        return jsonify({
            "command": matched,
            "description": cmd_info["desc"],
            "output": output,
            "return_code": result.returncode,
        })
    except subprocess.TimeoutExpired:
        return jsonify({"error": f"'{matched}' timed out after 30s"}), 504
    except Exception as e:
        logger.error(f"Command error ({matched}): {e}")
        return jsonify({"error": "Command execution failed"}), 500


@app.route("/api/commands", methods=["GET"])
def list_commands():
    """List all whitelisted commands."""
    return jsonify({
        "commands": [{"name": k, "description": v["desc"]} for k, v in ALLOWED_COMMANDS.items()]
    })


# ---------------------------------------------------------------------------
# Routes — file upload
# ---------------------------------------------------------------------------

@app.route("/api/upload", methods=["POST"])
def upload_file():
    """Upload a file for the voice agent (images, text, code, etc.)."""
    if "file" not in request.files:
        return jsonify({"error": "No file provided"}), 400

    file = request.files["file"]
    if not file.filename:
        return jsonify({"error": "No filename"}), 400

    allowed_exts = {
        ".png", ".jpg", ".jpeg", ".gif", ".webp",
        ".pdf", ".txt", ".md", ".json", ".csv",
        ".html", ".js", ".py", ".ts", ".css",
    }
    ext = Path(file.filename).suffix.lower()
    if ext not in allowed_exts:
        return jsonify({"error": f"File type '{ext}' not allowed"}), 400

    safe_name = re.sub(r"[^\w\-.]", "_", file.filename)[:80]
    save_name = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{safe_name}"
    save_path = UPLOADS_DIR / save_name
    file.save(save_path)

    is_image = ext in {".png", ".jpg", ".jpeg", ".gif", ".webp"}
    result = {
        "filename": save_name,
        "original_name": file.filename,
        "path": str(save_path),
        "type": "image" if is_image else "text",
        "size": save_path.stat().st_size,
        "url": f"/uploads/{save_name}",
    }
    if not is_image and ext != ".pdf":
        try:
            result["content_preview"] = save_path.read_text(encoding="utf-8", errors="replace")[:2000]
        except Exception:
            pass

    logger.info(f"Upload: {file.filename} → {save_path} ({result['size']} bytes)")
    return jsonify(result)


# GET /api/uploads (the uploads listing) lives in routes/static_files.py::list_uploads, behind
# the same uploads gate as the files themselves.


# ---------------------------------------------------------------------------
# Z.AI direct fallback — voice path (glm-5-flash)
# ---------------------------------------------------------------------------

def get_zai_direct_response(message: str, session_id: str = None) -> str:
    """Call Z.AI glm-5-flash directly for voice fallback (no tools, no gateway)."""
    import urllib.request as _urlreq
    import json as _json
    zai_key = os.getenv("ZAI_API_KEY", "")
    if not zai_key:
        logger.warning("get_zai_direct_response: ZAI_API_KEY not set")
        return None
    payload = _json.dumps({
        "model": "glm-5-flash",
        "messages": [{"role": "user", "content": message}],
        "max_tokens": 512,
        "temperature": 0.7,
    }).encode()
    req = _urlreq.Request(
        "https://api.z.ai/api/paas/v4/chat/completions",
        data=payload,
        headers={"Authorization": f"Bearer {zai_key}", "Content-Type": "application/json"},
    )
    try:
        with _urlreq.urlopen(req, timeout=30) as r:
            d = _json.load(r)
        msg = d["choices"][0]["message"]
        if msg.get("reasoning_content") and not msg.get("content"):
            return None  # thinking-only block — treat as empty
        return msg.get("content", "").strip() or None
    except Exception as exc:
        logger.error(f"get_zai_direct_response failed: {exc}")
        return None


# ---------------------------------------------------------------------------
# WebSocket — Gateway proxy (/ws/clawdbot)
# ---------------------------------------------------------------------------

from services.tts import generate_tts_b64 as _generate_tts_b64


def _tts_bytes(text: str) -> bytes:
    """Generate TTS audio bytes for the WebSocket proxy."""
    b64 = _generate_tts_b64(text, voice="M1")
    if b64 is None:
        raise RuntimeError("TTS generation returned no audio")
    return base64.b64decode(b64)


# ---------------------------------------------------------------------------
# Proactive push — agent-initiated messages to connected browsers
# ---------------------------------------------------------------------------
# Registry of live /ws/clawdbot browser sockets. When a subagent finishes
# after the originating HTTP stream closed, the gateway layer's orphan
# continuation (services/gateways/openclaw.py) collects the agent's follow-up
# and calls push_proactive_message() — closing the "I'll let you know when
# it's done" loop that previously never fired.

# ws -> outbound queue.Queue. flask_sock's ws.send() is NOT thread-safe, so
# proactive pushes (generated on the gateway daemon thread) must NOT call
# ws.send() directly while the socket's own bridge loop is also writing
# (keepalives, TTS) — interleaved frames corrupt the push channel (WS-2).
# Instead every producer ENQUEUES here; each /ws/clawdbot connection's own
# writer coroutine is the ONLY thread that touches its ws.send().
_push_clients_lock = threading.Lock()
_push_clients: dict = {}          # ws -> queue.Queue
_push_client_session: dict = {}   # ws -> last sessionKey the browser declared (WS-9)
_push_receipt_lock = threading.Lock()  # serialize JSONL appends (WS-10)


def _register_push_client(ws):
    """Register a browser socket and return its outbound queue."""
    q = queue.Queue()
    with _push_clients_lock:
        _push_clients[ws] = q
        n = len(_push_clients)
    logger.info(f"Proactive push: client registered ({n} connected)")
    return q


def _unregister_push_client(ws):
    with _push_clients_lock:
        q = _push_clients.pop(ws, None)
        _push_client_session.pop(ws, None)
    if q is not None:
        # Wake the socket's writer so it can observe the shutdown and exit.
        try:
            q.put_nowait(None)
        except Exception:
            pass


def _push_queue_for(ws):
    with _push_clients_lock:
        return _push_clients.get(ws)


def _note_push_client_session(ws, session_key):
    """Record the sessionKey a browser is working under so proactive pushes
    can be scoped to the right tab (WS-9)."""
    if not session_key:
        return
    with _push_clients_lock:
        if ws in _push_clients:
            _push_client_session[ws] = session_key


def push_proactive_message(text: str, session_key: str = None, run_id: str = None):
    """Broadcast an agent-initiated message to connected browsers.

    Called from a daemon thread (gateway loop side) — no Flask context.
    The raw text (with [CANVAS:...] etc. action tags intact) is sent so the
    client can dispatch the tags; TTS is generated from the tag-stripped text.

    Delivery is scoped (WS-9): if this frame carries a sessionKey AND a client
    has declared one, it is delivered only on a match; a client that never
    declared a sessionKey receives everything (backward compatible).

    Returns the number of browsers the message was enqueued to. Every call
    writes a receipt — including delivered=0 when no browser is connected, so
    a completion that finished while the owner was away leaves a trace (the
    gateway then relays it through the tenant's own agent).
    """
    with _push_clients_lock:
        any_clients = bool(_push_clients)
    if not any_clients:
        # Checked before TTS so an unheard push does not pay for audio.
        logger.warning("Proactive push: no connected browsers — message not delivered "
                       f"(text: {text[:100]})")
        _write_push_receipt(text, session_key, 0, 0, run_id)
        return 0
    spoken = re.sub(r'\[[A-Z_]+(?::[^\]]*)?\]', '', text).strip()
    audio_b64 = None
    if spoken:
        try:
            audio_b64 = _generate_tts_b64(spoken, voice="M1")
        except Exception as e:
            logger.warning(f"Proactive push: TTS failed ({e}) — sending text-only")
    frame = json.dumps({
        "type": "proactive_message",
        "text": text,
        "audio": audio_b64,
        "sessionKey": session_key,
        "ts": time.time(),
    })
    with _push_clients_lock:
        targets = list(_push_clients.items())
        sessions = dict(_push_client_session)
    if not targets:  # last browser left while TTS was generating
        logger.warning("Proactive push: no connected browsers — message not delivered "
                       f"(text: {text[:100]})")
        _write_push_receipt(text, session_key, 0, 0, run_id)
        return 0
    delivered = 0
    for ws, q in targets:
        reg_key = sessions.get(ws)
        if session_key and reg_key and reg_key != session_key:
            continue  # WS-9: this push belongs to a different session/tab
        try:
            q.put_nowait(frame)
            delivered += 1
        except Exception:
            _unregister_push_client(ws)
    logger.info(f"Proactive push: enqueued to {delivered}/{len(targets)} clients")
    _write_push_receipt(text, session_key, delivered, len(targets), run_id)
    return delivered


_PUSH_RECEIPT_PATH = Path("/app/runtime/transcripts/.proactive-push.jsonl")


def _write_push_receipt(text, session_key, delivered, clients, run_id=None):
    """Durable, host-visible receipt of every proactive push (bind-mounted
    transcripts dir) — drives the JamFlow board signal and survives restarts.
    Multiple proactive-push threads can append concurrently, so serialize the
    write under a module lock — interleaved partial lines corrupt the JSONL
    the JamFlow board reads (WS-10)."""
    try:
        receipt_path = _PUSH_RECEIPT_PATH
        line = json.dumps({
            "ts": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
            "session_key": session_key,
            "run_id": run_id,
            "delivered": delivered,
            "clients": clients,
            "text": text[:300],
        }) + "\n"
        with _push_receipt_lock:
            with open(receipt_path, "a") as f:
                f.write(line)
    except Exception as e:
        logger.debug(f"Proactive push receipt write failed: {e}")


# Register with the gateway layer so orphan subagent continuations can reach us.
try:
    from services.gateways.openclaw import set_proactive_handler
    set_proactive_handler(push_proactive_message)
    logger.info("Proactive push handler registered with gateway layer")
except Exception as _e:
    logger.error(f"Failed to register proactive push handler: {_e}")


def _warm_gateway_connection():
    """Open the persistent gateway WS at boot (it is otherwise lazy — first
    chat.send opens it). Without this, subagent completions that land after a
    server restart but before the first user message are invisible to the
    orphan-continuation watcher.

    Docker restarts a container's dependencies without compose `depends_on`
    ordering, so on a host reboot the openvoiceui container can come up
    before its openclaw gateway is listening. `_ensure_connected()` only
    makes 5 attempts before raising — without a retry here, the persistent
    WS never opens and stays down until a user turn happens to call
    `_ensure_connected()` lazily. So retry indefinitely in the background;
    once a lazy connect (or a later warm-up attempt) succeeds,
    `_ensure_connected()` short-circuits on `_connected` being True, so this
    loop's next pass is a cheap no-op and simply returns.
    """
    try:
        from services.gateway_manager import gateway_manager as _gm
        from services.warmup_retry import retry_until_success
        _gw = _gm.get('openclaw')
        if _gw is None or not _gw.is_configured():
            return
        _conn = _gw._router._get_connection(None)
        _conn._ensure_started()

        def _connect():
            asyncio.run_coroutine_threadsafe(
                _conn._ensure_connected(), _conn._loop).result(timeout=30)

        def _log(msg):
            logger.warning(f"Gateway boot warm-up {msg}")

        attempts = retry_until_success(_connect, time.sleep, _log)
        logger.info(f"Gateway WS warmed at boot after {attempts} attempt(s) — "
                    f"orphan completion watcher live")
    except Exception as e:
        logger.warning(f"Gateway boot warm-up failed permanently (will connect lazily): {e}")


threading.Thread(target=_warm_gateway_connection,
                 name='gateway-warmup', daemon=True).start()


@sock.route("/ws/clawdbot")
def clawdbot_websocket(ws):
    """WebSocket proxy between the frontend and the OpenClaw Gateway.

    Connects to CLAWDBOT_GATEWAY_URL, performs the protocol-3 handshake,
    then bridges messages bidirectionally — generating TTS audio for every
    assistant response before forwarding to the client.
    """
    # --- Clerk auth check (uses same cookie/header as HTTP routes) ---
    from services.auth import verify_clerk_token, get_token_from_request
    token = get_token_from_request()
    user_id = verify_clerk_token(token) if token else None
    if not user_id:
        logger.warning("WebSocket rejected — no valid Clerk token")
        ws.send(json.dumps({"type": "error", "message": "Unauthorized"}))
        ws.close()
        return
    logger.info(f"WebSocket authenticated: user_id={user_id}")

    gateway_url = os.getenv("CLAWDBOT_GATEWAY_URL", "ws://127.0.0.1:18791")
    auth_token = os.getenv("CLAWDBOT_AUTH_TOKEN")

    if not auth_token:
        logger.error("CLAWDBOT_AUTH_TOKEN not set — WebSocket rejected")
        ws.send(json.dumps({"type": "error", "message": "Server configuration error"}))
        ws.close()
        return

    # Register for proactive push (agent-initiated messages) for the life of
    # this socket — delivery is independent of the gateway bridge below.
    _register_push_client(ws)

    # Outbound queue for THIS socket. Every frame written to the browser —
    # gateway relays, keepalives, and proactive pushes — goes through here and
    # is drained by the single _writer() coroutine below, which is the only
    # thread that ever calls ws.send() (WS-2).
    out_q = _push_queue_for(ws)

    async def _run():
        loop = asyncio.get_running_loop()

        def _enqueue(frame):
            try:
                out_q.put_nowait(frame)
            except Exception:
                pass

        try:
            async with websockets.connect(gateway_url) as gw:
                logger.info(f"WebSocket connected to Gateway at {gateway_url}")

                # Handshake — these pre-writer sends are safe: no other thread
                # writes ws until _writer() starts (push only ENQUEUES).
                challenge = json.loads(await asyncio.wait_for(gw.recv(), timeout=10.0))
                logger.debug(f"Gateway challenge: {challenge.get('event')}")

                from services.gateways.compat import build_connect_params
                connect_params = build_connect_params(
                    auth_token=auth_token,
                    client_id="webchat",
                    client_mode="webchat",
                    platform="web",
                    user_agent="openvoice-ui-webchat/1.0.0",
                    caps=[],
                )
                await gw.send(json.dumps({
                    "type": "req",
                    "id": f"connect-{uuid.uuid4()}",
                    "method": "connect",
                    "params": connect_params,
                }))

                resp = json.loads(await asyncio.wait_for(gw.recv(), timeout=10.0))
                if resp.get("type") != "res" or not resp.get("ok"):
                    logger.error(f"Gateway handshake failed: {resp}")
                    ws.send(json.dumps({"type": "error", "message": "Gateway handshake failed"}))
                    ws.close()
                    return

                logger.info("Gateway handshake OK")
                _enqueue(json.dumps({"type": "connected", "message": "Connected to OpenClaw Gateway"}))

                async def _writer():
                    """The ONLY writer to ws. Drains the per-client outbound
                    queue so flask_sock's non-thread-safe send() is never
                    called from two threads (WS-2)."""
                    while True:
                        frame = await loop.run_in_executor(None, out_q.get)
                        if frame is None:  # shutdown sentinel
                            break
                        try:
                            ws.send(frame)
                        except Exception:
                            break

                async def _from_client():
                    while True:
                        # run_in_executor so blocking ws.receive() doesn't
                        # starve _from_gateway() while the socket is idle (WS-3).
                        msg = await loop.run_in_executor(None, ws.receive)
                        if not msg:
                            break
                        data = json.loads(msg)
                        if data.get("type") == "chat.send":
                            _note_push_client_session(ws, data.get("sessionKey"))
                            await gw.send(json.dumps({
                                "type": "req",
                                "id": f"chat-{uuid.uuid4()}",
                                "method": "chat.send",
                                "params": {
                                    "content": data.get("content", ""),
                                    "sessionKey": data.get("sessionKey", "main"),
                                },
                            }))

                async def _from_gateway():
                    while True:
                        try:
                            data = json.loads(await asyncio.wait_for(gw.recv(), timeout=25.0))
                        except asyncio.TimeoutError:
                            # Idle is normal — keep both legs alive. The browser
                            # socket doubles as the proactive-push channel, so it
                            # must survive long silences (nginx idle timeouts too).
                            _enqueue(json.dumps({"type": "keepalive", "ts": time.time()}))
                            continue
                        if data.get("type") != "event":
                            continue
                        event = data.get("event")
                        payload = data.get("payload", {})

                        if event == "agent.message":
                            content = payload.get("content", "")
                            if content:
                                try:
                                    # Offload blocking TTS so it doesn't stall
                                    # keepalives/frames on this loop (WS-5).
                                    raw = await loop.run_in_executor(None, _tts_bytes, content)
                                    audio_b64 = base64.b64encode(raw).decode()
                                    _enqueue(json.dumps({
                                        "type": "assistant_message",
                                        "text": content,
                                        "audio": audio_b64,
                                    }))
                                except Exception as e:
                                    logger.error(f"TTS failed in WebSocket handler: {e}")
                                    _enqueue(json.dumps({"type": "assistant_message", "text": content}))

                        elif event == "agent.stream.delta":
                            _enqueue(json.dumps({
                                "type": "text_delta",
                                "delta": payload.get("delta", ""),
                            }))

                        elif event == "agent.stream.end":
                            _enqueue(json.dumps({"type": "stream_end"}))

                # WS-8: stop as soon as ANY leg ends; cancel the survivors.
                # An executor-blocked ws.receive()/out_q.get() can't be
                # cancelled directly — the None sentinel unblocks the writer and
                # exiting the `async with` (closing gw) + ws.close() unblocks the
                # rest, so the upstream connection can't leak.
                tasks = {asyncio.ensure_future(_writer()),
                         asyncio.ensure_future(_from_client()),
                         asyncio.ensure_future(_from_gateway())}
                try:
                    await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                finally:
                    _enqueue(None)  # release the writer's blocking get
                    for t in tasks:
                        t.cancel()
                    try:
                        ws.close()
                    except Exception:
                        pass

        except (ConnectionRefusedError, OSError) as e:
            logger.error(f"Cannot reach Gateway at {gateway_url}: {e}")
            try:
                ws.send(json.dumps({"type": "error", "message": "Cannot connect to Gateway"}))
                ws.close()
            except Exception:
                pass
        except Exception as e:
            logger.error(f"WebSocket error: {e}")
            try:
                ws.send(json.dumps({"type": "error", "message": "Connection error"}))
                ws.close()
            except Exception:
                pass

    try:
        asyncio.run(_run())
    except Exception as e:
        logger.error(f"Fatal WebSocket error: {e}")
    finally:
        _unregister_push_client(ws)


# ---------------------------------------------------------------------------
# OpenClaw Control UI — WebSocket proxy
# ---------------------------------------------------------------------------
# Accepts browser WS (Clerk-authed via __session cookie), connects to the
# internal openclaw gateway, and relays all frames bidirectionally.
# Transparently injects the gateway auth token into the connect handshake
# so the user never has to enter it.

@sock.route("/openclaw-ui")
def openclaw_ui_websocket(ws):
    """WebSocket proxy for OpenClaw Control UI behind Clerk ADMIN auth."""
    from services.auth import verify_clerk_token, get_token_from_request, is_admin_user
    token = get_token_from_request()
    user_id = verify_clerk_token(token) if token else None
    if not user_id:
        logger.warning("OpenClaw UI WebSocket rejected — no valid Clerk token")
        ws.send(json.dumps({"type": "error", "message": "Unauthorized"}))
        ws.close()
        return
    # This proxy injects the OPERATOR-scope gateway token into the handshake, so
    # it must be admin-only — an allowlisted voice user must NOT get operator RPC
    # (SEC-5/WS-4). Being in ALLOWED_USER_IDS gates the voice app, not admin ops.
    if not is_admin_user(user_id):
        logger.warning(f"OpenClaw UI WebSocket rejected — non-admin user_id={user_id}")
        ws.send(json.dumps({"type": "error", "message": "Admin access required"}))
        ws.close()
        return
    logger.info(f"OpenClaw UI WebSocket authenticated (admin): user_id={user_id}")

    # Same RPC method allowlist the hardened HTTP admin proxy enforces
    # (routes/admin.py ALLOWED_RPC_METHODS) — the proxy relays operator-scope
    # frames, so arbitrary methods (config.patch / chat.send / sessions.*) must
    # not be reachable here. `connect` is permitted (handshake).
    from routes.admin import ALLOWED_RPC_METHODS
    _allowed_ws_methods = set(ALLOWED_RPC_METHODS) | {"connect"}

    gateway_url = os.getenv("CLAWDBOT_GATEWAY_URL", "ws://127.0.0.1:18791")
    auth_token = os.getenv("CLAWDBOT_AUTH_TOKEN")

    if not auth_token:
        logger.error("CLAWDBOT_AUTH_TOKEN not set — OpenClaw UI WebSocket rejected")
        ws.send(json.dumps({"type": "error", "message": "Server configuration error"}))
        ws.close()
        return

    async def _run():
        loop = asyncio.get_running_loop()
        try:
            async with websockets.connect(gateway_url, origin="http://localhost:18789") as gw:
                logger.info(f"OpenClaw UI: connected to Gateway at {gateway_url}")

                async def _from_client():
                    """Relay browser → openclaw, injecting auth token on connect."""
                    while True:
                        # run_in_executor so blocking ws.receive() doesn't
                        # starve _from_gateway() while idle (WS-3).
                        msg = await loop.run_in_executor(None, ws.receive)
                        if not msg:
                            break
                        try:
                            data = json.loads(msg)
                            if data.get("type") == "req":
                                _method = data.get("method")
                                # Enforce the RPC method allowlist — reject any
                                # method the HTTP admin proxy also refuses.
                                if _method and _method not in _allowed_ws_methods:
                                    logger.warning(
                                        f"OpenClaw UI: blocked disallowed RPC method {_method!r}"
                                    )
                                    ws.send(json.dumps({
                                        "type": "res",
                                        "id": data.get("id"),
                                        "error": {"message": f"Method not allowed: {_method}"},
                                    }))
                                    continue
                                if _method == "connect":
                                    if "params" not in data:
                                        data["params"] = {}
                                    data["params"]["auth"] = {"token": auth_token}
                                    msg = json.dumps(data)
                        except (json.JSONDecodeError, TypeError):
                            pass  # Non-JSON frame — relay as-is
                        await gw.send(msg)

                async def _from_gateway():
                    """Relay openclaw → browser (raw, no transformation).
                    This is the ONLY writer to ws in this bridge (the client
                    leg only relays to the gateway), so no per-client queue is
                    needed here."""
                    while True:
                        try:
                            msg = await asyncio.wait_for(gw.recv(), timeout=300.0)
                            ws.send(msg)
                        except asyncio.TimeoutError:
                            logger.warning("OpenClaw UI: gateway recv() timed out")
                            ws.send(json.dumps({"type": "error", "message": "Gateway timeout"}))
                            return

                # WS-8: end as soon as either leg finishes; cancel the survivor.
                # The executor-blocked ws.receive() can't be cancelled, but
                # exiting the `async with` (closing gw) + ws.close() unblocks it,
                # so the upstream gateway connection can't leak.
                tasks = {asyncio.ensure_future(_from_client()),
                         asyncio.ensure_future(_from_gateway())}
                try:
                    await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                finally:
                    for t in tasks:
                        t.cancel()
                    try:
                        ws.close()
                    except Exception:
                        pass

        except (ConnectionRefusedError, OSError) as e:
            logger.error(f"OpenClaw UI: cannot reach Gateway at {gateway_url}: {e}")
            ws.send(json.dumps({"type": "error", "message": "Cannot connect to Gateway"}))
            ws.close()
        except Exception as e:
            logger.error(f"OpenClaw UI WebSocket error: {e}")
            ws.send(json.dumps({"type": "error", "message": "Connection error"}))
            ws.close()

    try:
        asyncio.run(_run())
    except Exception as e:
        logger.error(f"OpenClaw UI: fatal WebSocket error: {e}")


# ---------------------------------------------------------------------------
# WebSocket — xAI Grok Realtime proxy (/ws/xai-realtime)
# ---------------------------------------------------------------------------
# Transparent bidirectional bridge: browser ↔ wss://api.x.ai/v1/realtime
# The XAI_API_KEY is injected as a Bearer header here — it never reaches
# the browser. All JSON event frames (including base64 PCM16 audio) are
# forwarded as-is; the adapter in xai-realtime.js handles the protocol.
# ---------------------------------------------------------------------------

@sock.route("/ws/xai-realtime")
def xai_realtime_websocket(ws):
    """WebSocket proxy between browser and xAI Grok Realtime API."""
    from services.auth import verify_clerk_token, get_token_from_request

    token   = get_token_from_request()
    user_id = verify_clerk_token(token) if token else None
    if not user_id:
        logger.warning("xAI Realtime WebSocket rejected — no valid Clerk token")
        ws.send(json.dumps({"type": "error", "message": "Unauthorized"}))
        ws.close()
        return

    api_key = os.getenv("XAI_API_KEY")
    if not api_key:
        logger.error("XAI_API_KEY not set — xAI Realtime WebSocket rejected")
        ws.send(json.dumps({"type": "error", "message": "xAI API key not configured on server"}))
        ws.close()
        return

    xai_url = "wss://api.x.ai/v1/realtime?model=grok-voice-think-fast-1.0"
    logger.info(f"xAI Realtime: new session for user_id={user_id}")

    async def _run():
        try:
            async with websockets.connect(
                xai_url,
                additional_headers={
                    "Authorization": f"Bearer {api_key}",
                    "OpenAI-Beta":   "realtime=v1",
                },
            ) as xai_ws:
                logger.info("xAI Realtime: upstream connection established")

                async def _from_client():
                    """Browser → xAI: relay all frames (text and binary)."""
                    loop = asyncio.get_running_loop()
                    while True:
                        # run_in_executor so blocking ws.receive() doesn't
                        # starve _from_xai() waiting on async xai_ws.recv().
                        msg = await loop.run_in_executor(None, ws.receive)
                        if msg is None:
                            logger.info("xAI Realtime: browser disconnected")
                            break
                        await xai_ws.send(msg)

                async def _from_xai():
                    """xAI → browser: relay all frames (text and binary)."""
                    while True:
                        try:
                            msg = await asyncio.wait_for(xai_ws.recv(), timeout=300.0)
                        except asyncio.TimeoutError:
                            logger.warning("xAI Realtime: upstream recv() timed out after 300s")
                            try:
                                ws.send(json.dumps({"type": "error", "message": "xAI connection timed out"}))
                            except Exception:
                                pass
                            return
                        # ws.send() handles both str and bytes frames
                        ws.send(msg)

                await asyncio.gather(_from_client(), _from_xai())

        except (ConnectionRefusedError, OSError) as e:
            logger.error(f"xAI Realtime: cannot connect to xAI: {e}")
            try:
                ws.send(json.dumps({"type": "error", "message": "Cannot connect to xAI API"}))
            except Exception:
                pass
        except Exception as e:
            logger.error(f"xAI Realtime: WebSocket error: {e}")
            try:
                ws.send(json.dumps({"type": "error", "message": "xAI Realtime connection error"}))
            except Exception:
                pass

    try:
        asyncio.run(_run())
    except Exception as e:
        logger.error(f"xAI Realtime: fatal error: {e}")
    finally:
        logger.info(f"xAI Realtime: session ended for user_id={user_id}")


# ---------------------------------------------------------------------------
# WebSocket — Co-browsing screencast proxy (/ws/browse-stream)  [#154 Phase 2]
# ---------------------------------------------------------------------------
# Bridges the browser's viewer canvas to the shared jambot-browse service's
# CDP screencast WS. Browser side is flask-sock; upstream is the aiohttp WS on
# the browse service. Frames flow down (base64 JPEG), user/viewer events flow
# up (Phase 3 consumes them; Phase 2 relays them through untouched). The
# service key + this tenant are injected server-side — never exposed to the
# browser. Tenant is fixed by this OVU instance's env, so a browser can only
# ever watch ITS OWN tenant's session.
@sock.route("/ws/browse-stream")
def browse_stream_websocket(ws):
    from services.auth import verify_clerk_token, get_token_from_request
    # Clerk gate unless auth disabled (single-user self-host), matching the app.
    # The real browser auto-sends the __session cookie (same origin). The internal
    # agent key is also accepted (via ?agent_key= or X-Agent-Key) for parity with
    # the /api/browse/* HTTP proxy, which lets internal tooling attach a viewer.
    # SEC-046: gate on the SAME "Clerk configured" flag services/auth.py uses (publishable key),
    # never on the secret. The app never needs CLERK_SECRET_KEY, so removing it from a tenant
    # must not switch this gate off (it used to: "publishable OR secret").
    from services.auth import _CLERK_CONFIGURED
    if _CLERK_CONFIGURED:
        _agent_key = os.getenv("AGENT_API_KEY", "").strip()
        _presented = request.args.get("agent_key", "") or request.headers.get("X-Agent-Key", "")
        _agent_ok = bool(_agent_key) and _presented == _agent_key
        token = get_token_from_request()
        if not _agent_ok and not (token and verify_clerk_token(token)):
            try:
                ws.send(json.dumps({"type": "error", "message": "Unauthorized"}))
            finally:
                ws.close()
            return

    svc_url = os.getenv("BROWSE_SERVICE_URL", "http://jambot-browse:8712").rstrip("/")
    svc_key = os.getenv("BROWSE_SERVICE_KEY", "").strip()
    tenant = (os.getenv("JAMBOT_TENANT") or os.getenv("TENANT_NAME") or "default").strip()
    ws_url = svc_url.replace("http://", "ws://").replace("https://", "wss://")
    stream_url = f"{ws_url}/session/stream?tenant={tenant}"
    if svc_key:
        stream_url += f"&key={svc_key}"

    async def _run():
        try:
            async with websockets.connect(stream_url, max_size=8 * 1024 * 1024) as up:
                logger.info("browse-stream connected upstream tenant=%s", tenant)

                async def _down():
                    """Upstream frames → browser."""
                    async for frame in up:
                        ws.send(frame)

                async def _up():
                    """Browser viewer events → upstream (Phase 3 acts on them)."""
                    loop = asyncio.get_running_loop()
                    while True:
                        msg = await loop.run_in_executor(None, ws.receive)
                        if msg is None:
                            break
                        await up.send(msg)

                done, pending = await asyncio.wait(
                    [asyncio.create_task(_down()), asyncio.create_task(_up())],
                    return_when=asyncio.FIRST_COMPLETED,
                )
                for t in pending:
                    t.cancel()
        except Exception as e:
            logger.info("browse-stream closed tenant=%s: %s", tenant, e)
            try:
                ws.send(json.dumps({"type": "error", "message": "browse stream ended"}))
            except Exception:
                pass

    try:
        asyncio.run(_run())
    finally:
        try:
            ws.close()
        except Exception:
            pass


# ---------------------------------------------------------------------------
# Terminal — PTY over WebSocket (backs the terminal canvas block)
# ---------------------------------------------------------------------------
# Spawns a login shell inside THIS container, cwd'd to the dedicated code
# sandbox (CODE_WORKSPACE_DIR). Gated by the Clerk __session cookie (same as
# /ws/clawdbot) OR the internal X-Agent-Key. Pure-Python pty — no node-pty.

@sock.route("/ws/terminal")
def terminal_websocket(ws):
    import pty, select, struct, fcntl, termios, signal, threading
    from services.auth import verify_clerk_token, get_token_from_request
    from services.paths import CODE_WORKSPACE_DIR

    # --- auth: valid Clerk session OR internal agent key ---
    token = get_token_from_request()
    user_id = verify_clerk_token(token) if token else None
    agent_key = os.getenv("AGENT_API_KEY", "").strip()
    has_agent_key = bool(agent_key) and request.headers.get("X-Agent-Key") == agent_key
    if not user_id and not has_agent_key:
        logger.warning("Terminal WS rejected — no valid Clerk token / agent key")
        try:
            ws.send(json.dumps({"type": "error", "message": "Unauthorized"}))
            ws.close()
        except Exception:
            pass
        return
    logger.info(f"Terminal WS authenticated: user={user_id or 'agent-key'}")

    try:
        CODE_WORKSPACE_DIR.mkdir(parents=True, exist_ok=True)
    except Exception:
        pass

    pid, master_fd = pty.fork()
    if pid == 0:
        # --- child: become the login shell ---
        os.environ["TERM"] = "xterm-256color"
        try:
            os.chdir(str(CODE_WORKSPACE_DIR))
        except Exception:
            pass
        shell = os.environ.get("SHELL") or "/bin/bash"
        try:
            os.execvp(shell, [shell, "-l"])
        except Exception:
            try:
                os.execvp("/bin/sh", ["/bin/sh"])
            except Exception:
                os._exit(1)

    # --- parent: bridge pty <-> websocket ---
    stop = threading.Event()

    def _pty_to_ws():
        while not stop.is_set():
            try:
                r, _, _ = select.select([master_fd], [], [], 0.2)
                if master_fd in r:
                    data = os.read(master_fd, 65536)
                    if not data:
                        break
                    ws.send(data.decode("utf-8", "replace"))
            except (OSError, ValueError):
                break
            except Exception:
                break
        stop.set()
        try:
            ws.close()
        except Exception:
            pass

    reader = threading.Thread(target=_pty_to_ws, daemon=True)
    reader.start()

    try:
        while not stop.is_set():
            msg = ws.receive()
            if msg is None:
                break
            try:
                data = json.loads(msg)
            except Exception:
                continue
            mtype = data.get("type")
            if mtype == "input":
                try:
                    os.write(master_fd, data.get("data", "").encode("utf-8"))
                except OSError:
                    break
            elif mtype == "resize":
                try:
                    cols = max(1, int(data.get("cols", 80)))
                    rows = max(1, int(data.get("rows", 24)))
                    fcntl.ioctl(master_fd, termios.TIOCSWINSZ,
                                struct.pack("HHHH", rows, cols, 0, 0))
                except Exception:
                    pass
    except Exception as e:
        logger.error(f"Terminal WS error: {e}")
    finally:
        stop.set()
        try:
            os.kill(pid, signal.SIGKILL)
        except Exception:
            pass
        try:
            os.close(master_fd)
        except Exception:
            pass
        try:
            os.waitpid(pid, 0)
        except Exception:
            pass


# ---------------------------------------------------------------------------
# Game Library API
# ---------------------------------------------------------------------------
# Serves game catalog data to canvas pages (same-origin, no CORS needed).
# Reads /app/runtime/game-catalog.json (mounted from /mnt/game-drive/catalog.json).
# Falls back to HTTP proxy at host:6360 if file not mounted.
_GAME_CATALOG_PATH = os.getenv("GAME_CATALOG_PATH", "/app/runtime/game-catalog.json")
_GAME_SERVER_URL = os.getenv("GAME_SERVER_URL", "http://172.19.0.1:6360")

def _load_game_catalog():
    """Load catalog from mounted file. Returns list or None on failure."""
    # Check primary path, then uploads fallback (uploads/ is always mounted)
    for path in [_GAME_CATALOG_PATH, "/app/runtime/uploads/game-catalog.json"]:
        try:
            with open(path) as f:
                return json.load(f)
        except Exception:
            continue
    return None

@app.route("/api/games", methods=["GET"])
def games_api():
    catalog = _load_game_catalog()
    if catalog is not None:
        system = request.args.get("system", "").lower()
        genre = request.args.get("genre", "").lower()
        search = request.args.get("search", "").lower()
        games = [g for g in catalog if g.get("status") == "downloaded"]
        if system:
            games = [g for g in games if g.get("system", "").lower() == system]
        if genre:
            games = [g for g in games if g.get("genre", "").lower() == genre]
        if search:
            games = [g for g in games if
                     search in g.get("title", "").lower() or
                     search in g.get("description", "").lower()]
        return jsonify(games)
    try:
        import requests as _req
        r = _req.get(f"{_GAME_SERVER_URL}/api/games", params=dict(request.args), timeout=10)
        return Response(r.content, status=r.status_code, mimetype="application/json")
    except Exception as e:
        return jsonify({"error": f"Game server unreachable: {e}"}), 502

@app.route("/api/games/stats", methods=["GET"])
def games_stats_api():
    catalog = _load_game_catalog()
    if catalog is not None:
        total = len(catalog)
        downloaded = sum(1 for g in catalog if g.get("status") == "downloaded")
        pending = sum(1 for g in catalog if g.get("status") == "pending")
        failed = sum(1 for g in catalog if g.get("status") == "failed")
        by_system, by_genre = {}, {}
        for g in catalog:
            if g.get("status") == "downloaded":
                s = g.get("system", "unknown")
                by_system[s] = by_system.get(s, 0) + 1
                gr = g.get("genre", "unknown")
                by_genre[gr] = by_genre.get(gr, 0) + 1
        return jsonify({"total": total, "downloaded": downloaded,
                        "pending": pending, "failed": failed,
                        "bySystem": by_system, "byGenre": by_genre})
    try:
        import requests as _req
        r = _req.get(f"{_GAME_SERVER_URL}/api/stats", timeout=10)
        return Response(r.content, status=r.status_code, mimetype="application/json")
    except Exception as e:
        return jsonify({"error": f"Game server unreachable: {e}"}), 502

@app.route("/api/games/systems", methods=["GET"])
def games_systems_api():
    catalog = _load_game_catalog()
    if catalog is not None:
        systems = sorted(set(g.get("system", "") for g in catalog if g.get("system")))
        return jsonify(systems)
    try:
        import requests as _req
        r = _req.get(f"{_GAME_SERVER_URL}/api/systems", timeout=10)
        return Response(r.content, status=r.status_code, mimetype="application/json")
    except Exception as e:
        return jsonify({"error": f"Game server unreachable: {e}"}), 502


# ---------------------------------------------------------------------------
# Per-endpoint rate limits (applied after all routes are registered)
# ---------------------------------------------------------------------------
# limiter.limit() returns a wrapped function — must assign it back into
# app.view_functions or the limit is silently discarded.
# The table lives in services/rate_limits.py (importable + tested).
from services.rate_limits import apply_endpoint_rate_limits
apply_endpoint_rate_limits(app)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    port = int(os.getenv("PORT", 5001))

    # Clean SIGTERM shutdown so systemd stop/restart works correctly.
    # Restart=on-failure only triggers on non-zero exit — os._exit(0) prevents that.
    def _handle_sigterm(signum, frame):
        logger.info("SIGTERM received — shutting down.")
        os._exit(0)

    signal.signal(signal.SIGTERM, _handle_sigterm)
    signal.signal(signal.SIGHUP, signal.SIG_IGN)

    try:
        from services.gateways.openclaw import OPENCLAW_TESTED_VERSION
        _oc_ver = OPENCLAW_TESTED_VERSION
    except ImportError:
        _oc_ver = "unknown"

    logger.info(f"OpenVoiceUI starting on port {port}")
    logger.info(f"  Frontend  → http://localhost:{port}/")
    logger.info(f"  Health    → http://localhost:{port}/health/ready")
    logger.info(f"  Admin     → http://localhost:{port}/src/admin.html")
    logger.info(f"  Gateway   → {os.getenv('CLAWDBOT_GATEWAY_URL', 'ws://127.0.0.1:18791')}")
    logger.info(f"  Tested OpenClaw version: {_oc_ver}")

    host = os.getenv("HOST", "127.0.0.1")  # Docker sets HOST=0.0.0.0; VPS stays loopback
    app.run(host=host, port=port, debug=False, threaded=True)