Upload 6 files

README.md

@@ -14,7 +14,7 @@ Deploy the contents of this `HF` folder at the root of your Hugging Face Space.
 
 This profile is CPU-first:
 
-- Normal TTS: Edge, Piper, Anime presets
+- Normal TTS: Edge cloud voices, curated Piper models, and Silero Russian voices
 - Voice clone: Pocket TTS CPU backend
 - GPU/ZeroGPU: not required
 
@@ -25,11 +25,33 @@ Set these in Space settings:
 - `VOICECRAFT_AUTH_MODE=license` - requires a live active license and registered device on every `/tts` request.
 - `LICENSE_VALIDATION_URL` - the deployed Google Apps Script URL. The current production URL is the code default, but setting it explicitly is recommended.
 - `ENABLE_CLONE_ENGINES=1` - keeps voice clone enabled.
-- `HF_TOKEN` - Hugging Face read token from an account that has accepted the Pocket TTS model terms.
-
-`API_SECRET` is legacy compatibility only. Do not put a shared API secret in distributed desktop apps. After every supported desktop build sends `X-License-Key` and `X-Device-ID`, remove the sheet `ApiSecret` value and rotate/remove the old Space `API_SECRET`.
-
-Optional modes:
+- `HF_TOKEN` - Hugging Face read token from an account that has accepted the Pocket TTS model terms.
+
+`API_SECRET` is legacy compatibility only. Do not put a shared API secret in distributed desktop apps. After every supported desktop build sends `X-License-Key` and `X-Device-ID`, remove the sheet `ApiSecret` value and rotate/remove the old Space `API_SECRET`.
+
+Instead of setting many Space secrets, you can set one protected secret named
+`VOICECRAFT_SPACE_CONFIG_JSON`. Individual secrets still work and override this
+combined config when both are present.
+
+```json
+{
+  "space_id": "razapro857-firebase",
+  "auth_mode": "license",
+  "max_concurrent_jobs": 1,
+  "service_version": "3.1.0",
+  "enable_clone_engines": true,
+  "hf_token": "hf_xxxxxxxxxxxxxxxxx",
+  "firebase_service_account": {
+    "type": "service_account",
+    "project_id": "voicecraft-1",
+    "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n",
+    "client_email": "firebase-adminsdk-xxx@voicecraft-1.iam.gserviceaccount.com",
+    "token_uri": "https://oauth2.googleapis.com/token"
+  }
+}
+```
+
+Optional modes:
 
 - `VOICECRAFT_AUTH_MODE=layered` requires both the live license/device and `API_SECRET`.
 - `VOICECRAFT_AUTH_MODE=api_secret` is legacy-only and is not recommended for public distribution.

app.py

@@ -1,4 +1,5 @@
-import os, io, asyncio, tempfile, threading, re, subprocess, shutil, logging, secrets, sys, platform, math, hashlib, time
+import os, io, asyncio, tempfile, threading, re, subprocess, shutil, logging, secrets, sys, platform, math, hashlib, time, json
+from contextlib import nullcontext
 
 try:
     import spaces
@@ -15,6 +16,85 @@ except ImportError:
 
     spaces = _SpacesFallback()
 
+try:
+    import firebase_admin
+    from firebase_admin import auth as fb_auth, firestore, credentials
+    HAS_FIREBASE = True
+except Exception:
+    HAS_FIREBASE = False
+    firebase_admin = None
+    fb_auth = None
+    firestore = None
+    credentials = None
+
+
+def _apply_space_config_secret():
+    """Allow one protected HF secret to populate the usual env settings."""
+    raw = os.environ.get("VOICECRAFT_SPACE_CONFIG_JSON", "").strip()
+    if not raw:
+        return
+    try:
+        config = json.loads(raw)
+        if not isinstance(config, dict):
+            raise ValueError("VOICECRAFT_SPACE_CONFIG_JSON must be a JSON object")
+    except Exception as exc:
+        print("Space config JSON notice:", exc)
+        return
+
+    def set_default(env_name, *keys, transform=str):
+        if os.environ.get(env_name):
+            return
+        for key in keys:
+            if key in config and config[key] not in (None, ""):
+                os.environ[env_name] = transform(config[key])
+                return
+
+    firebase_json = (
+        config.get("firebase_service_account_json")
+        or config.get("firebase_admin_json")
+        or config.get("firebase_service_account")
+        or config.get("service_account")
+    )
+    if firebase_json and not os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON"):
+        os.environ["FIREBASE_SERVICE_ACCOUNT_JSON"] = (
+            firebase_json if isinstance(firebase_json, str) else json.dumps(firebase_json)
+        )
+
+    set_default("VOICECRAFT_SPACE_ID", "space_id", "voicecraft_space_id")
+    set_default("VOICECRAFT_AUTH_MODE", "auth_mode", "voicecraft_auth_mode")
+    set_default("VOICECRAFT_MAX_CONCURRENT_JOBS", "max_concurrent_jobs", "voicecraft_max_concurrent_jobs")
+    set_default("VOICECRAFT_SERVICE_VERSION", "service_version", "voicecraft_service_version")
+    set_default("MIN_CLIENT_VERSION", "min_client_version", "minimum_version")
+    set_default("ENABLE_CLONE_ENGINES", "enable_clone_engines", "clone_enabled", transform=lambda value: "true" if bool(value) else "false")
+    set_default("HF_TOKEN", "hf_token", "huggingface_token")
+    set_default("VOICECRAFT_API_SECRET", "api_secret", "voicecraft_api_secret")
+
+
+_apply_space_config_secret()
+
+_fb_app = None
+_firestore_db = None
+def _init_firebase():
+    global _fb_app, _firestore_db
+    if not HAS_FIREBASE:
+        return None
+    if _fb_app is None:
+        try:
+            credential_json = os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON", "").strip()
+            cred_path = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', '').strip()
+            if credential_json:
+                cred = credentials.Certificate(json.loads(credential_json))
+                _fb_app = firebase_admin.initialize_app(cred)
+                _firestore_db = firestore.client()
+            elif cred_path and os.path.exists(cred_path):
+                cred = credentials.Certificate(cred_path)
+                _fb_app = firebase_admin.initialize_app(cred)
+                _firestore_db = firestore.client()
+        except Exception as e:
+            print("Firebase init notice:", e)
+            return None
+    return _firestore_db
+
 import gradio as gr
 import requests
 
@@ -24,17 +104,23 @@ from fastapi import Form, Request, HTTPException, UploadFile, File
 from fastapi.responses import StreamingResponse, JSONResponse
 from fastapi.middleware.gzip import GZipMiddleware
 
-app = gr.Server(
+SERVICE_VERSION = os.environ.get("VOICECRAFT_SERVICE_VERSION", "3.1.0").strip()
+
+app = gr.Server(
+    debug=False,
     title="VoiceCraft TTS Server",
-    description="VoiceCraft desktop API with CPU voice cloning support.",
-    version="2.0",
+    description="VoiceCraft desktop service.",
+    version=SERVICE_VERSION,
+    docs_url=None,
+    redoc_url=None,
+    openapi_url=None,
+    enable_monitoring=False,
 )
 app.add_middleware(GZipMiddleware, minimum_size=1000)
 
 
 # ══════════════════════════════════════════════════════════════════
-# SECURITY — server-side authorization
-# Production mode validates each license/device with Google Apps Script.
+# SECURITY — server-side Firebase authorization
 # ══════════════════════════════════════════════════════════════════
 _RAW_API_SECRET = os.environ.get("API_SECRET", "").strip()
 if _RAW_API_SECRET.startswith("hf_") and not (
@@ -49,10 +135,7 @@ API_SECRET = (
     or os.environ.get("APP_API_SECRET", "").strip()
     or ("" if _RAW_API_SECRET.startswith("hf_") else _RAW_API_SECRET)
 )
-DEFAULT_LICENSE_VALIDATION_URL = (
-    "https://script.google.com/macros/s/"
-    "AKfycbx6KWT18JHUevX9zXhzsOg40_Ek7wNDhpTIesQ63Lm6aPNCoxhujQL8z_Ll9Dt6-cQ/exec"
-)
+DEFAULT_LICENSE_VALIDATION_URL = os.environ.get("DEFAULT_LICENSE_VALIDATION_URL", "").strip()
 LICENSE_VALIDATION_URL = (
     os.environ.get("LICENSE_VALIDATION_URL", "").strip()
     or DEFAULT_LICENSE_VALIDATION_URL
@@ -62,11 +145,19 @@ if AUTH_MODE not in {"license", "layered", "api_secret"}:
     AUTH_MODE = "license"
 MIN_CLIENT_VERSION = os.environ.get("MIN_CLIENT_VERSION", "3.0.0").strip()
 MAX_CLONE_CHARACTERS = 60_000
-CLONE_CHUNK_CHARACTERS = 1800
+CLONE_CHUNK_CHARACTERS = 1800
+CLONE_RETRY_CHUNK_CHARACTERS = 900
 CLONE_ENGINE_PREFIXES = ("f5tts:",)
 CLONE_ENGINES = ["f5tts"]
 CLONE_BACKEND_NAME = "pocket-tts-cpu"
-BASE_ENGINES = ["edge", "piper", "silero", "anime"]
+BASE_ENGINES = ["edge", "piper", "silero"]
+SPACE_ID = os.environ.get("VOICECRAFT_SPACE_ID", "").strip() or os.environ.get("SPACE_ID", "").strip() or "unregistered"
+MAX_CONCURRENT_JOBS = max(1, min(int(os.environ.get("VOICECRAFT_MAX_CONCURRENT_JOBS", "1")), 20))
+_SPACE_CAPACITY = threading.BoundedSemaphore(MAX_CONCURRENT_JOBS)
+_SPACE_STATE_LOCK = threading.Lock()
+_ACTIVE_JOBS = 0
+_POLICY_LOCK = threading.Lock()
+_POLICY_CACHE = {"loaded_at": 0.0, "maintenance_mode": False, "minimum_version": MIN_CLIENT_VERSION}
 
 
 def _env_flag(name: str, default=None):
@@ -76,16 +167,162 @@ def _env_flag(name: str, default=None):
     return raw.strip().lower() in {"1", "true", "yes", "on", "active", "enabled"}
 
 
-def clone_engines_enabled() -> bool:
-    explicit = _env_flag("ENABLE_CLONE_ENGINES", None)
-    if explicit is not None:
-        return explicit
-    # CPU clone is now the default backend. Set ENABLE_CLONE_ENGINES=0 for TTS-only Spaces.
-    return True
-
-
-def available_engines():
-    return BASE_ENGINES + (CLONE_ENGINES if clone_engines_enabled() else [])
+def clone_engines_enabled() -> bool:
+    explicit = _env_flag("ENABLE_CLONE_ENGINES", None)
+    if explicit is not None:
+        return explicit
+    # CPU clone is now the default backend. Set ENABLE_CLONE_ENGINES=0 for TTS-only Spaces.
+    return True
+
+
+def clone_backend_available() -> bool:
+    if not clone_engines_enabled():
+        return False
+    if _POCKET_MODEL_ERROR or _POCKET_GENERATION_ERROR:
+        return False
+    if _POCKET_MODEL is not None:
+        return bool(getattr(_POCKET_MODEL, "has_voice_cloning", False))
+    return _hf_token_configured()
+
+
+def clone_backend_status() -> dict:
+    if not clone_engines_enabled():
+        return {"enabled": False, "ready": False, "message": "disabled"}
+    if _POCKET_GENERATION_ERROR:
+        return {"enabled": False, "ready": False, "message": "generation_failed"}
+    if _POCKET_MODEL is not None:
+        ready = bool(getattr(_POCKET_MODEL, "has_voice_cloning", False))
+        return {
+            "enabled": ready,
+            "ready": ready,
+            "message": "ready" if ready else "model_loaded_without_clone_weights",
+        }
+    if _POCKET_MODEL_ERROR:
+        return {"enabled": False, "ready": False, "message": "model_startup_failed"}
+    if not _hf_token_configured():
+        return {"enabled": False, "ready": False, "message": "hf_token_required"}
+    return {"enabled": True, "ready": False, "message": "loading"}
+
+
+def available_engines():
+    return BASE_ENGINES + (CLONE_ENGINES if clone_backend_available() else [])
+
+
+def get_runtime_policy():
+    now = time.monotonic()
+    with _POLICY_LOCK:
+        if now - float(_POLICY_CACHE.get("loaded_at", 0.0)) < 30:
+            return dict(_POLICY_CACHE)
+    policy = {
+        "loaded_at": now,
+        "maintenance_mode": False,
+        "minimum_version": MIN_CLIENT_VERSION,
+        "space_enabled": True,
+        "space_clone_enabled": clone_backend_available(),
+    }
+    db = _init_firebase()
+    if db is not None:
+        try:
+            runtime_doc = db.collection("public_config").document("runtime").get()
+            if runtime_doc.exists:
+                runtime = runtime_doc.to_dict() or {}
+                policy["maintenance_mode"] = bool(runtime.get("maintenance_mode", False))
+                policy["minimum_version"] = str(runtime.get("minimum_version") or MIN_CLIENT_VERSION)
+            if SPACE_ID != "unregistered":
+                space_doc = db.collection("spaces").document(SPACE_ID).get()
+                if space_doc.exists:
+                    space_data = space_doc.to_dict() or {}
+                    policy["space_enabled"] = bool(space_data.get("enabled", False))
+                    policy["space_clone_enabled"] = bool(space_data.get("clone_enabled", False)) and clone_backend_available()
+        except Exception:
+            pass
+    with _POLICY_LOCK:
+        _POLICY_CACHE.clear()
+        _POLICY_CACHE.update(policy)
+    return dict(policy)
+
+
+def _clean_hf_space_url(value):
+    raw = str(value or "").strip().rstrip("/")
+    if not raw.startswith("https://"):
+        return ""
+    host = raw.split("/", 3)[2].lower()
+    if host == "hf.space" or host.endswith(".hf.space") or host == "huggingface.co" or host.endswith(".huggingface.co"):
+        return raw
+    return ""
+
+
+def _space_public_payload(snapshot_id, item):
+    url = _clean_hf_space_url((item or {}).get("url"))
+    if not url:
+        return None
+    return {
+        "id": str(snapshot_id or "")[:80],
+        "name": str((item or {}).get("name") or snapshot_id or "")[:80],
+        "url": url,
+        "clone_enabled": bool((item or {}).get("clone_enabled", False)),
+        "priority": int((item or {}).get("priority", 100) or 100),
+        "max_concurrent_jobs": max(1, int((item or {}).get("max_concurrent_jobs", 1) or 1)),
+    }
+
+
+def _license_space_pool(db, uid, user_data, license_key, lic):
+    """Return only the Spaces this signed-in user may see/use."""
+    if db is None:
+        return []
+
+    assigned_ids = set()
+    assigned_urls = set()
+    for source in (user_data or {}, lic or {}):
+        for key in (
+            "space_ids",
+            "clone_space_ids",
+            "dedicated_space_ids",
+            "assigned_space_ids",
+        ):
+            for value in source.get(key) or []:
+                if value:
+                    assigned_ids.add(str(value).strip())
+        for key in (
+            "spaces",
+            "clone_spaces",
+            "dedicated_spaces",
+            "assigned_spaces",
+        ):
+            for value in source.get(key) or []:
+                if isinstance(value, dict):
+                    payload = _space_public_payload(value.get("id") or value.get("name"), value)
+                    if payload:
+                        assigned_urls.add(payload["url"])
+                else:
+                    url = _clean_hf_space_url(value)
+                    if url:
+                        assigned_urls.add(url)
+
+    dedicated = []
+    auto_pool = []
+    try:
+        for snapshot in db.collection("spaces").where("enabled", "==", True).stream():
+            item = snapshot.to_dict() or {}
+            payload = _space_public_payload(snapshot.id, item)
+            if not payload:
+                continue
+            is_dedicated = (
+                snapshot.id in assigned_ids
+                or payload["url"] in assigned_urls
+                or str(item.get("assigned_uid") or "").strip() == str(uid or "").strip()
+                or str(item.get("assigned_license_key") or item.get("assigned_license") or "").strip() == str(license_key or "").strip()
+            )
+            if is_dedicated:
+                dedicated.append(payload)
+            elif not item.get("assigned_uid") and not item.get("assigned_license_key") and not item.get("assigned_license"):
+                auto_pool.append(payload)
+    except Exception:
+        return []
+
+    rows = dedicated or auto_pool
+    rows.sort(key=lambda item: (item["priority"], item["name"].lower()))
+    return rows
 
 
 def verify_token(request: Request):
@@ -97,88 +334,127 @@ def verify_token(request: Request):
         raise HTTPException(status_code=403, detail="Unauthorized")
 
 
-_LICENSE_CACHE = {}
-_LICENSE_CACHE_LOCK = threading.Lock()
 _LICENSE_RATE_STATE = {}
 _LICENSE_RATE_LOCK = threading.Lock()
 
-
 def _bounded_env_int(name: str, default: int, minimum: int, maximum: int) -> int:
     try:
         return max(minimum, min(maximum, int(os.environ.get(name, default))))
     except (TypeError, ValueError):
         return default
 
-
-def _license_cache_key(license_key: str, device_id: str) -> str:
-    raw = f"{license_key}\0{device_id}".encode("utf-8")
-    return hashlib.sha256(raw).hexdigest()
-
-
-def _license_flag(value) -> bool:
-    return str(value or "").strip().lower() in {
-        "active", "enabled", "enable", "on", "true", "yes", "1"
-    }
-
-
-def _version_tuple(value: str) -> tuple[int, ...]:
-    parts = re.findall(r"\d+", str(value or ""))
-    numbers = [int(part) for part in parts[:4]]
-    return tuple((numbers + [0, 0, 0, 0])[:4])
-
-
-def _validate_license_sync(license_key: str, device_id: str) -> dict:
-    cache_key = _license_cache_key(license_key, device_id)
-    now = time.monotonic()
-    with _LICENSE_CACHE_LOCK:
-        cached = _LICENSE_CACHE.get(cache_key)
-        if cached and cached["expires_at"] > now:
-            return cached["payload"]
+def _version_tuple(value: str) -> tuple[int, ...]:
+    parts = re.findall(r"\d+", str(value or ""))
+    numbers = [int(part) for part in parts[:4]]
+    return tuple((numbers + [0, 0, 0, 0])[:4])
+
+
+def _reserve_monthly_usage(db, uid, license_key, lic, is_clone, requested_characters):
+    from datetime import datetime, timezone
+
+    requested_characters = max(0, int(requested_characters or 0))
+    if not requested_characters:
+        return
+    month_id = datetime.now(timezone.utc).strftime("%Y%m")
+    usage_ref = db.collection("usage_monthly").document(f"{uid}_{month_id}")
+    used_field = "clone_characters" if is_clone else "tts_characters"
+    reserved_field = "clone_reserved" if is_clone else "tts_reserved"
+    limit = int(
+        lic.get("monthly_clone_characters" if is_clone else "monthly_characters")
+        or 0
+    )
+    transaction = db.transaction()
+
+    @firestore.transactional
+    def reserve(transaction):
+        snapshot = usage_ref.get(transaction=transaction)
+        usage = snapshot.to_dict() if snapshot.exists else {}
+        used = max(0, int(usage.get(used_field) or 0))
+        reserved = max(0, int(usage.get(reserved_field) or 0))
+        if limit and used + reserved + requested_characters > limit:
+            label = "voice-clone" if is_clone else "TTS character"
+            raise HTTPException(
+                status_code=403,
+                detail=f"Monthly {label} allowance has been reached",
+            )
+        transaction.set(
+            usage_ref,
+            {
+                "uid": uid,
+                "license_key": license_key,
+                "period": month_id,
+                reserved_field: reserved + requested_characters,
+                "updated_at": firestore.SERVER_TIMESTAMP,
+            },
+            merge=True,
+        )
+
+    try:
+        reserve(transaction)
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logging.warning("Usage reservation failed: %s", exc.__class__.__name__)
+        raise HTTPException(503, "Usage allowance could not be verified")
+
+
+def verify_firebase_auth(request, is_clone: bool = False, requested_characters: int = 0):
+    if not HAS_FIREBASE or not fb_auth:
+        raise HTTPException(503, "Firebase authentication is not configured")
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        raise HTTPException(401, "Missing authentication token")
+    token = auth_header.split("Bearer ", 1)[1]
     try:
-        response = requests.get(
-            LICENSE_VALIDATION_URL,
-            params={
-                "action": "check",
-                "license_key": license_key,
-                "pc_user": "VoiceCraftDesktop",
-                "device_id": device_id,
-            },
-            headers={"User-Agent": "VoiceCraft-Server/3.0"},
-            timeout=(6, 20),
-        )
-        response.raise_for_status()
-        payload = response.json()
-    except Exception as exc:
-        logging.warning("License validation service unavailable: %s", exc.__class__.__name__)
-        raise HTTPException(status_code=503, detail="License validation service unavailable")
-    status = str(payload.get("status", "") or "").strip().lower()
-    if status not in {"active", "trial"}:
-        message = str(payload.get("message", "") or "").strip()
-        if status == "device_limit":
-            detail = message or "Maximum device limit reached for this license"
-        elif status:
-            detail = message or f"License is {status}"
-        else:
-            detail = "License is not active"
-        raise HTTPException(status_code=403, detail=detail)
-    ttl = _bounded_env_int("LICENSE_CACHE_TTL_SECONDS", 45, 5, 300)
-    with _LICENSE_CACHE_LOCK:
-        _LICENSE_CACHE[cache_key] = {
-            "payload": payload,
-            "expires_at": now + ttl,
-        }
-        if len(_LICENSE_CACHE) > 2048:
-            expired = [
-                key for key, value in _LICENSE_CACHE.items()
-                if value["expires_at"] <= now
-            ]
-            for key in expired:
-                _LICENSE_CACHE.pop(key, None)
-    return payload
-
-
-def _enforce_rate_limit(license_key: str, device_id: str, is_clone: bool):
-    identity = _license_cache_key(license_key, device_id)
+        decoded = fb_auth.verify_id_token(token)
+    except Exception:
+        raise HTTPException(401, "Invalid or expired authentication token")
+    uid = decoded["uid"]
+    db = _init_firebase()
+    if db is None:
+        raise HTTPException(503, "Firebase database is unavailable")
+    user_doc = db.collection("users").document(uid).get()
+    if not user_doc.exists:
+        raise HTTPException(403, "User account not found")
+    user_data = user_doc.to_dict()
+    if user_data.get("is_blocked"):
+        raise HTTPException(403, "Account has been blocked")
+    license_key = user_data.get("license_key")
+    if not license_key:
+        raise HTTPException(403, "No active license. Please activate a license key.")
+    license_doc = db.collection("licenses").document(license_key).get()
+    if not license_doc.exists:
+        raise HTTPException(403, "License key not found")
+    lic = license_doc.to_dict()
+    if lic.get("status") not in ("active",):
+        raise HTTPException(403, f"License is {lic.get('status', 'invalid')}")
+    device_id = str(request.headers.get("X-Device-ID", "")).strip()[:128]
+    device_ids = list(user_data.get("device_ids") or [])
+    if not device_id or device_id not in device_ids:
+        raise HTTPException(403, "This device is not registered for the license")
+    if len(device_ids) > max(1, int(lic.get("max_devices") or 1)):
+        raise HTTPException(403, "Maximum licensed devices exceeded")
+    
+    from datetime import datetime, timezone
+    if lic.get("expiry_date") and lic["expiry_date"].replace(tzinfo=timezone.utc) < datetime.now(timezone.utc):
+        raise HTTPException(403, "License has expired")
+
+    if is_clone and (
+        not lic.get("voice_clone")
+        or not user_data.get("voice_clone_enabled", lic.get("voice_clone", False))
+    ):
+        raise HTTPException(403, "Voice cloning not included in your plan")
+    _reserve_monthly_usage(
+        db,
+        uid,
+        license_key,
+        lic,
+        is_clone,
+        requested_characters,
+    )
+    return uid, license_key, lic, user_data
+
+def _enforce_rate_limit(uid: str, is_clone: bool):
     now = time.monotonic()
     window = 60.0
     limit = _bounded_env_int(
@@ -187,7 +463,7 @@ def _enforce_rate_limit(license_key: str, device_id: str, is_clone: bool):
         1,
         600,
     )
-    state_key = f"{identity}:{'clone' if is_clone else 'tts'}"
+    state_key = f"{uid}:{'clone' if is_clone else 'tts'}"
     with _LICENSE_RATE_LOCK:
         requests_in_window = [
             timestamp
@@ -199,27 +475,129 @@ def _enforce_rate_limit(license_key: str, device_id: str, is_clone: bool):
         requests_in_window.append(now)
         _LICENSE_RATE_STATE[state_key] = requests_in_window
 
-
-async def authorize_request(request: Request, engine: str) -> dict:
+async def authorize_request(request: Request, engine: str, requested_characters: int = 0) -> dict:
+    policy = await asyncio.to_thread(get_runtime_policy)
+    if policy.get("maintenance_mode"):
+        raise HTTPException(status_code=503, detail="VoiceCraft is temporarily under maintenance")
+    if not policy.get("space_enabled", True):
+        raise HTTPException(status_code=503, detail="This worker has been disabled")
+    if engine in CLONE_ENGINES and not policy.get("space_clone_enabled", False):
+        raise HTTPException(status_code=403, detail="Voice cloning is not available on this server")
     if AUTH_MODE in {"api_secret", "layered"}:
         verify_token(request)
-    if AUTH_MODE == "api_secret":
-        return {}
-    license_key = request.headers.get("X-License-Key", "").strip()
-    device_id = request.headers.get("X-Device-ID", "").strip()
-    client_version = request.headers.get("X-Client-Version", "").strip()
-    if _version_tuple(client_version) < _version_tuple(MIN_CLIENT_VERSION):
-        raise HTTPException(status_code=426, detail="VoiceCraft update required")
-    if not license_key or not device_id:
-        raise HTTPException(status_code=401, detail="License credentials are required")
-    if len(license_key) > 256 or len(device_id) > 128:
-        raise HTTPException(status_code=400, detail="Invalid license credentials")
-    payload = await asyncio.to_thread(_validate_license_sync, license_key, device_id)
+    if AUTH_MODE == "api_secret":
+        return {"uid": "", "license_key": "", "license": {}}
+        
+    client_version = request.headers.get("X-Client-Version", "").strip()
+    minimum_version = str(policy.get("minimum_version") or MIN_CLIENT_VERSION)
+    if not client_version:
+        raise HTTPException(status_code=426, detail="VoiceCraft client version is required")
+    if _version_tuple(client_version) < _version_tuple(minimum_version):
+        raise HTTPException(status_code=426, detail="VoiceCraft update required")
+        
     is_clone = engine in CLONE_ENGINES
-    if is_clone and not _license_flag(payload.get("voiceClone")):
-        raise HTTPException(status_code=403, detail="Voice cloning is not enabled for this license")
-    _enforce_rate_limit(license_key, device_id, is_clone)
-    return payload
+    uid, license_key, lic, _user = await asyncio.to_thread(
+        verify_firebase_auth,
+        request,
+        is_clone,
+        requested_characters,
+    )
+    _enforce_rate_limit(uid, is_clone)
+    return {"uid": uid, "license_key": license_key, "license": lic}
+
+
+def acquire_worker_slot():
+    global _ACTIVE_JOBS
+    if not _SPACE_CAPACITY.acquire(blocking=False):
+        raise HTTPException(status_code=503, detail="Worker is busy; try another Space")
+    with _SPACE_STATE_LOCK:
+        _ACTIVE_JOBS += 1
+        active = _ACTIVE_JOBS
+    db = _init_firebase()
+    if db is not None and SPACE_ID != "unregistered":
+        try:
+            db.collection("spaces").document(SPACE_ID).set({
+                "active_jobs": active,
+                "last_heartbeat": firestore.SERVER_TIMESTAMP,
+            }, merge=True)
+        except Exception:
+            pass
+
+
+def release_worker_slot():
+    global _ACTIVE_JOBS
+    with _SPACE_STATE_LOCK:
+        _ACTIVE_JOBS = max(0, _ACTIVE_JOBS - 1)
+        active = _ACTIVE_JOBS
+    _SPACE_CAPACITY.release()
+    db = _init_firebase()
+    if db is not None and SPACE_ID != "unregistered":
+        try:
+            db.collection("spaces").document(SPACE_ID).set({
+                "active_jobs": active,
+                "last_heartbeat": firestore.SERVER_TIMESTAMP,
+            }, merge=True)
+        except Exception:
+            pass
+
+
+def record_usage(auth_context, engine, characters, status, duration_ms):
+    uid = str((auth_context or {}).get("uid") or "")
+    if not uid:
+        return
+    db = _init_firebase()
+    if db is None:
+        return
+    is_clone = engine in CLONE_ENGINES
+    event = {
+        "uid": uid,
+        "license_key": str(auth_context.get("license_key") or ""),
+        "space_id": SPACE_ID,
+        "engine": engine,
+        "characters": int(characters),
+        "is_clone": is_clone,
+        "status": status,
+        "duration_ms": int(duration_ms),
+        "created_at": firestore.SERVER_TIMESTAMP,
+    }
+    try:
+        batch = db.batch()
+        event_ref = db.collection("usage_events").document()
+        user_ref = db.collection("users").document(uid)
+        space_ref = db.collection("spaces").document(SPACE_ID)
+        from datetime import datetime, timezone
+        month_id = datetime.now(timezone.utc).strftime("%Y%m")
+        monthly_ref = db.collection("usage_monthly").document(f"{uid}_{month_id}")
+        batch.set(event_ref, event)
+        billed_characters = int(characters) if status == "success" else 0
+        reserved_field = "clone_reserved" if is_clone else "tts_reserved"
+        user_updates = {
+            "generation_count": firestore.Increment(1),
+            "last_generation_at": firestore.SERVER_TIMESTAMP,
+            "tts_characters": firestore.Increment(0 if is_clone else billed_characters),
+            "clone_characters": firestore.Increment(billed_characters if is_clone else 0),
+        }
+        batch.set(user_ref, user_updates, merge=True)
+        batch.set(monthly_ref, {
+            "uid": uid,
+            "period": month_id,
+            "tts_characters": firestore.Increment(0 if is_clone else billed_characters),
+            "clone_characters": firestore.Increment(billed_characters if is_clone else 0),
+            reserved_field: firestore.Increment(-int(characters)),
+            "generation_count": firestore.Increment(1),
+            "updated_at": firestore.SERVER_TIMESTAMP,
+        }, merge=True)
+        if SPACE_ID != "unregistered":
+            batch.set(space_ref, {
+                "total_requests": firestore.Increment(1),
+                "total_characters": firestore.Increment(billed_characters),
+                "last_heartbeat": firestore.SERVER_TIMESTAMP,
+            }, merge=True)
+        batch.commit()
+    except Exception as exc:
+        logging.warning("Usage logging failed: %s", exc.__class__.__name__)
+
+
 
 # ══════════════════════════════════════════════════════════════════
 # PIPER TTS SETUP — Auto download on first run
@@ -325,46 +703,7 @@ def download_piper_model(voice_code: str) -> tuple:
     return onnx_path, json_path
 
 
-def download_piper_dynamic(voice_code: str) -> tuple:
-    """Dynamic Piper voice download — code format: piper:ar_JO-kareem-low"""
-    model_name = voice_code.replace("piper:", "")  # ar_JO-kareem-low
-    # Prevent path traversal
-    if ".." in model_name or "/" in model_name or "\\" in model_name:
-        raise ValueError(f"Invalid voice code: {voice_code}")
-    onnx_file = f"{model_name}.onnx"
-    json_file = f"{model_name}.onnx.json"
-    onnx_path = os.path.join(PIPER_MODELS_DIR, onnx_file)
-    json_path = os.path.join(PIPER_MODELS_DIR, json_file)
-
-    if os.path.exists(onnx_path) and os.path.exists(json_path):
-        return onnx_path, json_path
-
-    import urllib.request
-    # Model format: lang_code-voice_name-quality (e.g., ar_JO-kareem-low)
-    # HF path: ar/ar_JO/kareem/low/ar_JO-kareem-low.onnx
-    dash_parts = model_name.rsplit("-", 2)  # Split from right: ["ar_JO", "kareem", "low"]
-    if len(dash_parts) >= 3:
-        lang_code = dash_parts[0]    # ar_JO
-        voice = dash_parts[1]        # kareem
-        quality = dash_parts[2]      # low
-        lang = lang_code.split("_")[0]  # ar
-        hf_path = f"{lang}/{lang_code}/{voice}/{quality}"
-    else:
-        hf_path = f"{model_name}/{model_name}"
-
-    for fname, fpath in [(onnx_file, onnx_path), (json_file, json_path)]:
-        if not os.path.exists(fpath):
-            url = f"{PIPER_BASE_URL}/{hf_path}/{fname}"
-            print(f"📥 Downloading dynamic Piper model: {fname}")
-            try:
-                urllib.request.urlretrieve(url, fpath)
-            except Exception as e:
-                print(f"⚠️ Dynamic Piper download failed: {e}")
-                raise
-    return onnx_path, json_path
-
-
-def _piper_synth_chunk(text: str, onnx_path: str, json_path: str, length_scale: float) -> bytes:
+def _piper_synth_chunk(text: str, onnx_path: str, json_path: str, length_scale: float) -> bytes:
     with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as out_f:
         out_path = out_f.name
     try:
@@ -391,51 +730,97 @@ def _piper_synth_chunk(text: str, onnx_path: str, json_path: str, length_scale:
             os.unlink(out_path)
 
 
-def _ffmpeg_concat_wav(parts: list) -> bytes:
-    """Merge multiple WAV byte chunks using ffmpeg concat (same codec)."""
+def _numpy_concat(parts: list, is_mp3: bool) -> bytes:
     if len(parts) == 1:
         return parts[0]
-    tmp_files, concat_list, out_path = [], None, None
-    try:
-        for data in parts:
-            fd, path = tempfile.mkstemp(suffix=".wav")
-            os.close(fd)
-            with open(path, "wb") as f:
-                f.write(data)
-            tmp_files.append(path)
-        fd, concat_list = tempfile.mkstemp(suffix=".txt")
+    import numpy as np
+    import soundfile as sf
+    import subprocess, tempfile, os
+
+    audio_arrays = []
+    samplerate = 24000
+
+    for data in parts:
+        fd, in_path = tempfile.mkstemp(suffix=".mp3" if is_mp3 else ".wav")
         os.close(fd)
-        with open(concat_list, "w") as f:
-            for p in tmp_files:
-                f.write(f"file '{p}'\n")
-        fd, out_path = tempfile.mkstemp(suffix=".wav")
+        with open(in_path, "wb") as f:
+            f.write(data)
+        
+        wav_path = in_path
+        if is_mp3:
+            fd, wav_path = tempfile.mkstemp(suffix=".wav")
+            os.close(fd)
+            subprocess.run(["ffmpeg", "-y", "-i", in_path, wav_path], check=True, capture_output=True)
+            
+        try:
+            arr, sr = sf.read(wav_path)
+            audio_arrays.append(arr)
+            samplerate = sr
+        except Exception:
+            pass
+            
+        try: os.unlink(in_path)
+        except: pass
+        if is_mp3:
+            try: os.unlink(wav_path)
+            except: pass
+
+    if not audio_arrays:
+        return b""
+
+    result = audio_arrays[0]
+    fade_len = int(samplerate * 0.075)
+    silence_len = int(samplerate * 0.15)
+    silence = np.zeros(silence_len, dtype=result.dtype)
+
+    for next_arr in audio_arrays[1:]:
+        result = np.concatenate([result, silence])
+        if len(result) > fade_len and len(next_arr) > fade_len:
+            fade_out = np.linspace(1.0, 0.0, fade_len)
+            fade_in = np.linspace(0.0, 1.0, fade_len)
+            
+            if len(result.shape) > 1:
+                fade_out = fade_out[:, np.newaxis]
+                fade_in = fade_in[:, np.newaxis]
+                
+            overlap_result = result[-fade_len:] * fade_out
+            overlap_next = next_arr[:fade_len] * fade_in
+            
+            result[-fade_len:] = overlap_result + overlap_next
+            result = np.concatenate([result, next_arr[fade_len:]])
+        else:
+            result = np.concatenate([result, next_arr])
+
+    fd, out_wav = tempfile.mkstemp(suffix=".wav")
+    os.close(fd)
+    sf.write(out_wav, result, samplerate)
+
+    if is_mp3:
+        fd, out_mp3 = tempfile.mkstemp(suffix=".mp3")
         os.close(fd)
-        subprocess.run(["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
-                        "-f", "concat", "-safe", "0", "-i", concat_list,
-                        "-c", "copy", out_path], check=True, timeout=60, creationflags=CREATE_NO_WINDOW)
-        with open(out_path, "rb") as f:
-            return f.read()
-    except Exception:
-        return b"".join(parts)
-    finally:
-        for p in tmp_files:
-            try: os.unlink(p)
-            except Exception: pass
-        for x in (concat_list, out_path):
-            if x:
-                try: os.unlink(x)
-                except Exception: pass
+        subprocess.run(["ffmpeg", "-y", "-i", out_wav, out_mp3], check=True, capture_output=True)
+        with open(out_mp3, "rb") as f:
+            res = f.read()
+        os.unlink(out_mp3)
+        os.unlink(out_wav)
+        return res
+    else:
+        with open(out_wav, "rb") as f:
+            res = f.read()
+        os.unlink(out_wav)
+        return res
+
+def _ffmpeg_concat_wav(parts: list) -> bytes:
+    return _numpy_concat(parts, False)
 
 
 def synthesize_piper(text: str, voice_code: str, speed: float = 1.0) -> bytes:
     if not PIPER_READY:
         raise Exception("Piper is not available on this system")
-    # Pehle PIPER_VOICES dict mein check karo, nahi mila to dynamic download
-    if voice_code in PIPER_VOICES:
-        onnx_path, json_path = download_piper_model(voice_code)
-    else:
-        # Dynamic voice — direct HuggingFace se download
-        onnx_path, json_path = download_piper_dynamic(voice_code)
+    # Only explicitly tested models are permitted.
+    if voice_code not in PIPER_VOICES:
+        raise ValueError("Unsupported Piper voice")
+    onnx_path, json_path = download_piper_model(voice_code)
     length_scale = 1.0 / max(0.25, min(4.0, speed))
     # Lambi text ko chunk karo (Piper stdin limit + timeout avoid karne ke liye)
     if len(text) > 1400:
@@ -453,13 +838,21 @@ def synthesize_piper(text: str, voice_code: str, speed: float = 1.0) -> bytes:
     return _ffmpeg_concat_wav(parts)
 
 
+def sanitize_text(text: str) -> str:
+    import re
+    text = re.sub(r'[\u200B-\u200D\uFEFF]', '', text)
+    text = re.sub(r'[\x00-\x08\x0b-\x1f\x7f]', '', text)
+    text = re.sub(r'[ \t]+', ' ', text)
+    text = re.sub(r'\n\s*\n+', '\n\n', text)
+    return text.strip()
+
 def split_text(text: str, max_chars: int = 1400) -> list:
     """Text ko chunklara bol"""
-    text = text.strip()
+    text = sanitize_text(text)
     if not text:
         return []
     sentence_re = re.compile(
-        r'(?:(?<=[.!?\u0964\u06D4\u061F\u2026])\s+)|(?<=[\u3002\uff01\uff1f])'
+        r'(?<=[.!?\u0964\u06D4\u061F\u2026\u3002\uff01\uff1f])\s+(?=\S)'
     )
     chunks, current = [], ""
     for para in re.split(r'\n+', text):
@@ -602,48 +995,13 @@ def synthesize_silero(text: str, voice_code: str) -> bytes:
 
 
 def _ffmpeg_concat_mp3(parts: list) -> bytes:
-    """Merge multiple MP3 byte chunks using ffmpeg concat."""
-    if len(parts) == 1:
-        return parts[0]
-    tmp_files = []
-    concat_list = None
-    out_path = None
-    try:
-        for data in parts:
-            fd, path = tempfile.mkstemp(suffix=".mp3")
-            os.close(fd)
-            with open(path, "wb") as f:
-                f.write(data)
-            tmp_files.append(path)
-        fd, concat_list = tempfile.mkstemp(suffix=".txt")
-        os.close(fd)
-        with open(concat_list, "w") as f:
-            for p in tmp_files:
-                f.write(f"file '{p}'\n")
-        fd, out_path = tempfile.mkstemp(suffix=".mp3")
-        os.close(fd)
-        subprocess.run(["ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
-                        "-f", "concat", "-safe", "0", "-i", concat_list,
-                        "-c", "copy", out_path], check=True, timeout=60, creationflags=CREATE_NO_WINDOW)
-        with open(out_path, "rb") as f:
-            return f.read()
-    except Exception:
-        return b"".join(parts)
-    finally:
-        for p in tmp_files:
-            try: os.unlink(p)
-            except Exception: pass
-        if concat_list:
-            try: os.unlink(concat_list)
-            except Exception: pass
-        if out_path:
-            try: os.unlink(out_path)
-            except Exception: pass
+    return _numpy_concat(parts, True)
 
 
 async def _edge_synth_chunk(chunk: str, voice: str, kwargs: dict) -> bytes:
     """One chunk ki audio lao, 3 baar retry karo. Fail par b"" return."""
     import edge_tts
+    import asyncio
     for attempt in range(3):
         data = bytearray()
         try:
@@ -655,8 +1013,18 @@ async def _edge_synth_chunk(chunk: str, voice: str, kwargs: dict) -> bytes:
                 return bytes(data)
         except Exception:
             if attempt == 2:
+                if voice != "en-US-AriaNeural":
+                    try:
+                        comm = edge_tts.Communicate(chunk, "en-US-AriaNeural", **kwargs)
+                        async for packet in comm.stream():
+                            if packet["type"] == "audio" and packet.get("data"):
+                                data.extend(packet["data"])
+                        if data:
+                            return bytes(data)
+                    except Exception:
+                        pass
                 return b""
-            await asyncio.sleep(1 + attempt)
+            await asyncio.sleep(2)
     return b""
 
 
@@ -701,9 +1069,10 @@ CLONE_MODELS = {
     "f5tts:v1_base": ("f5tts", "multilingual", "Voice Clone"),
 }
 
-_POCKET_MODEL = None
-_POCKET_MODEL_ERROR = None
-_POCKET_MODEL_LOCK = threading.Lock()
+_POCKET_MODEL = None
+_POCKET_MODEL_ERROR = None
+_POCKET_GENERATION_ERROR = None
+_POCKET_MODEL_LOCK = threading.Lock()
 _POCKET_INFER_LOCK = threading.Lock()
 _POCKET_RESULT_CACHE = {}
 _POCKET_RESULT_CACHE_ORDER = []
@@ -712,8 +1081,9 @@ _POCKET_RESULT_CACHE_LIMIT = 8
 _POCKET_RESULT_CACHE_MAX_BYTES = 16 * 1024 * 1024
 _POCKET_VOICE_STATE_CACHE = {}
 _POCKET_VOICE_STATE_ORDER = []
-_POCKET_VOICE_STATE_LOCK = threading.Lock()
-_POCKET_VOICE_STATE_LIMIT = 6
+_POCKET_VOICE_STATE_LOCK = threading.Lock()
+_POCKET_VOICE_STATE_LIMIT = 6
+_POCKET_TORCH_PATCHED = False
 
 
 def _hf_token_configured() -> bool:
@@ -731,20 +1101,42 @@ def _pocket_clone_auth_error() -> str:
     )
 
 
-def _pocket_clone_ready() -> bool:
-    return bool(_POCKET_MODEL is not None and getattr(_POCKET_MODEL, "has_voice_cloning", False))
-
-
-def _load_pocket_model():
-    global _POCKET_MODEL, _POCKET_MODEL_ERROR
-    if _POCKET_MODEL is not None:
-        return _POCKET_MODEL
-    with _POCKET_MODEL_LOCK:
-        if _POCKET_MODEL is not None:
-            return _POCKET_MODEL
-        try:
-            from pocket_tts import TTSModel
-            _POCKET_MODEL = TTSModel.load_model()
+def _pocket_clone_ready() -> bool:
+    return bool(_POCKET_MODEL is not None and getattr(_POCKET_MODEL, "has_voice_cloning", False))
+
+
+def _patch_torch_for_pocket_tts():
+    """Pocket TTS mutates a streaming KV cache; no_grad keeps tensors mutable."""
+    global _POCKET_TORCH_PATCHED
+    if _POCKET_TORCH_PATCHED:
+        return
+    try:
+        import torch
+        torch.inference_mode = torch.no_grad
+        _POCKET_TORCH_PATCHED = True
+        logging.info("Pocket TTS torch compatibility patch enabled")
+    except Exception as exc:
+        logging.warning("Pocket TTS torch patch skipped: %s", exc.__class__.__name__)
+
+
+def _load_pocket_model():
+    global _POCKET_MODEL, _POCKET_MODEL_ERROR
+    if _POCKET_MODEL is not None:
+        return _POCKET_MODEL
+    with _POCKET_MODEL_LOCK:
+        if _POCKET_MODEL is not None:
+            return _POCKET_MODEL
+        try:
+            _patch_torch_for_pocket_tts()
+            from pocket_tts import TTSModel
+            _POCKET_MODEL = TTSModel.load_model()
+            # Pocket TTS is inference-only in this service. Disabling gradient
+            # tracking reduces CPU work and memory without changing synthesis.
+            try:
+                import torch
+                torch.set_grad_enabled(False)
+            except Exception:
+                pass
             _POCKET_MODEL_ERROR = None
             if getattr(_POCKET_MODEL, "has_voice_cloning", False):
                 print("Pocket TTS CPU voice clone ready")
@@ -822,6 +1214,43 @@ def _normalize_clone_reference(reference_path: str) -> str:
                 "Reference audio could not be decoded. "
                 f"Use a clear WAV, MP3, M4A, OGG, or FLAC file. {detail[:160]}"
             )
+        
+        try:
+            import soundfile as sf
+            import numpy as np
+            data, sr = sf.read(normalized_path)
+            if len(data.shape) > 1:
+                data = data.mean(axis=1)
+            
+            sr = sr or 24000
+            duration = len(data) / float(sr)
+            
+            # Auto-trim if longer than 12s, auto-tile if shorter than 3s
+            if duration > 12.0:
+                data = data[:int(12.0 * sr)]
+            elif duration < 3.0 and duration > 0.3:
+                repeats = int(np.ceil(3.0 / duration))
+                data = np.tile(data, repeats)[:int(3.0 * sr)]
+            
+            rms = np.sqrt(np.mean(data**2))
+            if rms < 0.0005:
+                raise ValueError("Reference audio is mostly silence. Please record a clearer voice sample.")
+            
+            target_rms = 10 ** (-20 / 20)
+            data = data * (target_rms / (rms + 1e-9))
+            
+            threshold = 10 ** (-40 / 20)
+            mask = np.abs(data) > threshold
+            if np.any(mask):
+                start = np.argmax(mask)
+                end = len(mask) - np.argmax(mask[::-1])
+                data = data[start:end]
+            
+            sf.write(normalized_path, data, sr)
+        except ValueError:
+            raise
+        except Exception:
+            pass
         return normalized_path
     except Exception:
         try:
@@ -838,7 +1267,14 @@ def _clone_request_key(text: str, reference_key: str) -> str:
 
 def _get_cached_clone(cache_key: str):
     with _POCKET_RESULT_CACHE_LOCK:
-        return _POCKET_RESULT_CACHE.get(cache_key)
+        audio = _POCKET_RESULT_CACHE.get(cache_key)
+        if audio is not None:
+            try:
+                _POCKET_RESULT_CACHE_ORDER.remove(cache_key)
+            except ValueError:
+                pass
+            _POCKET_RESULT_CACHE_ORDER.append(cache_key)
+        return audio
 
 
 def _cache_clone(cache_key: str, audio: bytes):
@@ -856,10 +1292,17 @@ def _cache_clone(cache_key: str, audio: bytes):
 
 def _get_cached_voice_state(reference_key: str):
     with _POCKET_VOICE_STATE_LOCK:
-        return _POCKET_VOICE_STATE_CACHE.get(reference_key)
+        voice_state = _POCKET_VOICE_STATE_CACHE.get(reference_key)
+        if voice_state is not None:
+            try:
+                _POCKET_VOICE_STATE_ORDER.remove(reference_key)
+            except ValueError:
+                pass
+            _POCKET_VOICE_STATE_ORDER.append(reference_key)
+        return voice_state
 
 
-def _cache_voice_state(reference_key: str, voice_state):
+def _cache_voice_state(reference_key: str, voice_state):
     with _POCKET_VOICE_STATE_LOCK:
         if reference_key in _POCKET_VOICE_STATE_CACHE:
             _POCKET_VOICE_STATE_ORDER.remove(reference_key)
@@ -867,11 +1310,27 @@ def _cache_voice_state(reference_key: str, voice_state):
         _POCKET_VOICE_STATE_ORDER.append(reference_key)
         while len(_POCKET_VOICE_STATE_ORDER) > _POCKET_VOICE_STATE_LIMIT:
             oldest = _POCKET_VOICE_STATE_ORDER.pop(0)
-            _POCKET_VOICE_STATE_CACHE.pop(oldest, None)
-
-
-@app.api(name="voicecraft_clone_backend", api_visibility="private", concurrency_limit=1)
-def _run_pocket_clone_cpu(text: str, reference_path: str, reference_key: str) -> bytes:
+            _POCKET_VOICE_STATE_CACHE.pop(oldest, None)
+
+
+def _estimate_min_clone_seconds(text: str) -> float:
+    words = len(re.findall(r"\S+", str(text or "")))
+    # A very conservative floor; natural speech can be fast, but a much shorter
+    # result usually means the model stopped early and skipped words.
+    return max(0.45, min(20.0, words * 0.105))
+
+
+def _generate_pocket_chunk(model, voice_state, chunk: str):
+    import numpy as np
+
+    audio = model.generate_audio(voice_state, chunk)
+    if hasattr(audio, "detach"):
+        audio = audio.detach().cpu().numpy()
+    return np.asarray(audio).squeeze().reshape(-1)
+
+
+@app.api(name="voicecraft_clone_backend", api_visibility="private", concurrency_limit=1)
+def _run_pocket_clone_cpu(text: str, reference_path: str, reference_key: str) -> bytes:
     if not reference_path or not os.path.exists(reference_path):
         raise ValueError("A reference audio file is required for voice cloning")
     model = _load_pocket_model()
@@ -892,13 +1351,36 @@ def _run_pocket_clone_cpu(text: str, reference_path: str, reference_key: str) ->
                 raise
             _cache_voice_state(reference_key, voice_state)
         chunks = split_text(text, max_chars=CLONE_CHUNK_CHARACTERS)
-        for chunk in chunks:
-            audio = model.generate_audio(voice_state, chunk)
-            if hasattr(audio, "detach"):
-                audio = audio.detach().cpu().numpy()
-            audio_np = np.asarray(audio).squeeze().reshape(-1)
-            if audio_np.size:
-                audio_parts.append(audio_np)
+        try:
+            import torch
+            inference_context = torch.no_grad()
+        except Exception:
+            inference_context = nullcontext()
+        with inference_context:
+            for chunk in chunks:
+                audio_np = _generate_pocket_chunk(model, voice_state, chunk)
+                duration = audio_np.size / float(getattr(model, "sample_rate", 24000) or 24000)
+                if (
+                    audio_np.size
+                    and len(chunk) > CLONE_RETRY_CHUNK_CHARACTERS
+                    and duration < _estimate_min_clone_seconds(chunk)
+                ):
+                    logging.warning("Clone chunk looked truncated; retrying with smaller chunks")
+                    smaller_parts = []
+                    for sub_chunk in split_text(chunk, max_chars=CLONE_RETRY_CHUNK_CHARACTERS):
+                        sub_audio = _generate_pocket_chunk(model, voice_state, sub_chunk)
+                        if sub_audio.size:
+                            smaller_parts.append(sub_audio)
+                    if smaller_parts:
+                        silence = np.zeros(int(model.sample_rate * 0.08), dtype=smaller_parts[0].dtype)
+                        joined = []
+                        for index, part in enumerate(smaller_parts):
+                            if index:
+                                joined.append(silence)
+                            joined.append(part)
+                        audio_np = np.concatenate(joined)
+                if audio_np.size:
+                    audio_parts.append(audio_np)
     if not audio_parts:
         raise RuntimeError("Voice clone model did not produce audio")
     if len(audio_parts) == 1:
@@ -919,58 +1401,100 @@ def _run_pocket_clone_cpu(text: str, reference_path: str, reference_key: str) ->
     return buf.getvalue()
 
 
-async def synthesize_f5tts(text: str, reference_path: str = None) -> bytes:
-    if not reference_path or not os.path.exists(reference_path):
-        raise ValueError("A reference audio file is required for voice cloning")
-    reference_key = await asyncio.to_thread(_reference_audio_key, reference_path)
-    cache_key = _clone_request_key(text, reference_key)
-    cached = _get_cached_clone(cache_key)
-    if cached is not None:
-        return cached
-    if _get_cached_voice_state(reference_key) is not None:
-        # Same reference voice is already encoded. Skip ffmpeg normalization for repeated
-        # desktop batches/generations; audio quality is unchanged because voice_state is reused.
-        audio = await asyncio.to_thread(
-            _run_pocket_clone_cpu, text, reference_path, reference_key
+async def synthesize_f5tts(text: str, reference_path: str = None) -> bytes:
+    global _POCKET_GENERATION_ERROR
+    if not reference_path or not os.path.exists(reference_path):
+        raise ValueError("A reference audio file is required for voice cloning")
+    try:
+        reference_key = await asyncio.to_thread(_reference_audio_key, reference_path)
+        cache_key = _clone_request_key(text, reference_key)
+        cached = _get_cached_clone(cache_key)
+        if cached is not None:
+            return cached
+        if _get_cached_voice_state(reference_key) is not None:
+            # Same reference voice is already encoded. Skip ffmpeg normalization for repeated
+            # desktop batches/generations; audio quality is unchanged because voice_state is reused.
+            audio = await asyncio.to_thread(
+                _run_pocket_clone_cpu, text, reference_path, reference_key
+            )
+        else:
+            normalized_path = await asyncio.to_thread(_normalize_clone_reference, reference_path)
+            try:
+                audio = await asyncio.to_thread(
+                    _run_pocket_clone_cpu, text, normalized_path, reference_key
+                )
+            finally:
+                try:
+                    os.unlink(normalized_path)
+                except OSError:
+                    pass
+        _cache_clone(cache_key, audio)
+        _POCKET_GENERATION_ERROR = None
+        return audio
+    except Exception as exc:
+        _POCKET_GENERATION_ERROR = exc.__class__.__name__
+        logging.exception("Voice clone generation failed")
+        raise RuntimeError("Voice cloning failed on this server. Please contact support.") from exc
+
+def post_process_bytes(audio: bytes, media_type: str) -> bytes:
+    """Apply gentle, smooth speech cleanup without block-gate clicks.
+
+    A single ffmpeg pass is faster than decoding into Python/SciPy, walking
+    every 20 ms window, writing a WAV, and encoding again. The conservative
+    denoise amount removes low-level hiss while preserving voice character.
+    """
+    suffix = ".mp3" if "mpeg" in media_type else ".wav"
+    fd, in_path = tempfile.mkstemp(suffix=suffix)
+    os.close(fd)
+    with open(in_path, "wb") as f:
+        f.write(audio)
+    fd, out_path = tempfile.mkstemp(suffix=suffix)
+    os.close(fd)
+    try:
+        ffmpeg = shutil.which("ffmpeg")
+        if not ffmpeg:
+            return audio
+        filters = (
+            "highpass=f=55,"
+            "afftdn=nr=4:nf=-55:tn=1,"
+            "alimiter=limit=0.891:attack=5:release=50"
         )
-    else:
-        normalized_path = await asyncio.to_thread(_normalize_clone_reference, reference_path)
-        try:
-            audio = await asyncio.to_thread(
-                _run_pocket_clone_cpu, text, normalized_path, reference_key
-            )
-        finally:
-            try:
-                os.unlink(normalized_path)
-            except OSError:
-                pass
-    _cache_clone(cache_key, audio)
-    return audio
-
-async def synthesize_anime(text: str, voice_code: str, style: str = None) -> bytes:
-    """Anime-style TTS using reliable Edge voices until a custom anime backend is configured."""
-    voice_map = {
-        "anime:en_whisper": ("en-US-AvaMultilingualNeural", "whispering", "+8%", "+4%", "+4Hz"),
-        "anime:ja_edge": ("ja-JP-NanamiNeural", "cheerful", "+10%", "+5%", "+5Hz"),
-        "anime:zh_piper": ("zh-CN-XiaoxiaoNeural", "cheerful", "+10%", "+5%", "+5Hz"),
-    }
-    edge_voice, fallback_style, rate, volume, pitch = voice_map.get(
-        voice_code,
-        ("en-US-AvaMultilingualNeural", "cheerful", "+10%", "+5%", "+5Hz"),
-    )
-    return await synthesize_edge(
-        text,
-        edge_voice,
-        rate=rate,
-        volume=volume,
-        pitch=pitch,
-        style=style or fallback_style,
-    )
+        cmd = [
+            ffmpeg, "-y", "-hide_banner", "-loglevel", "error",
+            "-i", in_path, "-vn", "-af", filters,
+        ]
+        if suffix == ".mp3":
+            # Explicit high-quality encoding avoids ffmpeg's lower default
+            # bitrate and prevents an avoidable quality drop.
+            cmd += ["-codec:a", "libmp3lame", "-b:a", "320k", out_path]
+        else:
+            cmd += ["-acodec", "pcm_s16le", out_path]
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            timeout=300,
+            creationflags=CREATE_NO_WINDOW,
+        )
+        if result.returncode != 0 or not os.path.exists(out_path) or os.path.getsize(out_path) <= 44:
+            logging.warning("Audio cleanup failed with ffmpeg exit code %s", result.returncode)
+            return audio
+        with open(out_path, "rb") as output_file:
+            return output_file.read()
+    except Exception as exc:
+        logging.warning("Audio cleanup failed: %s", exc.__class__.__name__)
+        return audio
+    finally:
+        for p in (in_path, out_path):
+            if os.path.exists(p):
+                try:
+                    os.unlink(p)
+                except OSError:
+                    pass
 
 @app.post("/tts")
 async def tts_endpoint(
     request: Request,
-    engine:  str = Form(...),   # "edge" | "piper" | "silero" | "anime" | "f5tts"
+    engine:  str = Form(...),   # "edge" | "piper" | "silero" | "f5tts"
     text:    str = Form(...),
     voice:   str = Form("en-US-AvaNeural"),   # edge voice code OR piper/silero code
     rate:    str = Form("+0%"),               # edge only
@@ -979,13 +1503,14 @@ async def tts_endpoint(
     speed:   float = Form(1.0),              # piper only
     style:   str = Form(None),               # edge style (emotion)
     styledegree: str = Form(None),           # edge style degree 0-2
+    post_process: bool = Form(True),         # audio post-processing toggle
     voice_cloning: bool = Form(False),       # voice cloning toggle
     reference_audio: UploadFile = File(None),
 ):
     if not text or not text.strip():
         return JSONResponse(status_code=400, content={"error": "Text is empty"})
 
-    text = text.strip()
+    text = text.strip()
 
     # Abuse / timeout guard — lambi text Chapter Mode se bhejo
     if len(text) > 60000:
@@ -994,14 +1519,35 @@ async def tts_endpoint(
             content={"error": "Text too long (max 60000 chars). Use Chapter Mode for longer text."},
         )
 
-    engine = engine.strip().lower()
-    await authorize_request(request, engine)
-
-    if engine in CLONE_ENGINES and not clone_engines_enabled():
-        return JSONResponse(
-            status_code=403,
-            content={"error": "Clone engines are disabled on this Space"},
-        )
+    engine = engine.strip().lower()
+    if engine not in available_engines():
+        return JSONResponse(status_code=400, content={"error": "Unsupported TTS engine"})
+
+    has_reference = bool(reference_audio is not None and reference_audio.filename)
+    if has_reference and engine not in CLONE_ENGINES:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Reference audio is only accepted for voice cloning"},
+        )
+    if engine in CLONE_ENGINES and not has_reference:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Voice cloning requires reference audio"},
+        )
+    if engine == "piper" and voice not in PIPER_VOICES:
+        return JSONResponse(status_code=400, content={"error": "Unsupported Piper voice"})
+    if engine == "silero" and not str(voice).startswith("silero:ru_"):
+        return JSONResponse(status_code=400, content={"error": "Unsupported Silero voice"})
+    if engine == "edge" and not re.match(r"^[a-z]{2,3}-[A-Z]{2}-.+Neural$", str(voice)):
+        return JSONResponse(status_code=400, content={"error": "Unsupported Edge voice"})
+
+    auth_context = await authorize_request(request, engine, len(text))
+
+    if engine in CLONE_ENGINES and not clone_backend_available():
+        return JSONResponse(
+            status_code=403,
+            content={"error": "Voice cloning is not available on this server"},
+        )
     if engine in CLONE_ENGINES and len(text) > MAX_CLONE_CHARACTERS:
         return JSONResponse(
             status_code=413,
@@ -1013,9 +1559,14 @@ async def tts_endpoint(
             },
         )
 
-    reference_path = None
-    try:
-        if reference_audio is not None and reference_audio.filename:
+    reference_path = None
+    started_at = time.monotonic()
+    generation_status = "error"
+    slot_acquired = False
+    try:
+        acquire_worker_slot()
+        slot_acquired = True
+        if has_reference:
             suffix = os.path.splitext(reference_audio.filename)[1].lower() or ".wav"
             if suffix not in (".wav", ".mp3", ".m4a", ".ogg", ".flac"):
                 return JSONResponse(status_code=400, content={"error": "Unsupported reference audio format"})
@@ -1052,15 +1603,17 @@ async def tts_endpoint(
             media = "audio/wav"
             fname = "tts_f5_clone.wav"
 
-        elif engine == "anime":
-            audio = await synthesize_anime(text, voice, style)
-            media = "audio/mpeg"
-            fname = "tts_anime.mp3"
-
         else:
             return JSONResponse(status_code=400, content={"error": f"Unknown engine: {engine}"})
 
-        return StreamingResponse(
+        # Preserve clone identity and improve speed: Pocket TTS already emits
+        # clean PCM. Re-filtering cloned audio can subtly change timbre and
+        # adds a full ffmpeg pass. Normal TTS still receives gentle cleanup.
+        if post_process and engine not in CLONE_ENGINES:
+            audio = await asyncio.to_thread(post_process_bytes, audio, media)
+
+        generation_status = "success"
+        return StreamingResponse(
             io.BytesIO(audio),
             media_type=media,
             headers={"Content-Disposition": f"attachment; filename={fname}"},
@@ -1070,53 +1623,288 @@ async def tts_endpoint(
         logging.error(f"TTS error: {e}", exc_info=True)
         return JSONResponse(
             status_code=500,
-            content={"error": f"Synthesis failed: {str(e)[:240]}"},
+            content={"error": "Audio generation failed. Please retry shortly."},
         )
-    finally:
-        try:
+    finally:
+        await asyncio.to_thread(
+            record_usage,
+            auth_context,
+            engine,
+            len(text),
+            generation_status,
+            int((time.monotonic() - started_at) * 1000),
+        )
+        if slot_acquired:
+            release_worker_slot()
+        try:
             if reference_path and os.path.exists(reference_path):
                 os.unlink(reference_path)
         except Exception:
             pass
 
 
-@app.get("/status")
-def status():
+
+from fastapi import Body
+
+@app.post("/register_user")
+async def register_user(request: Request, body: dict = Body(...)):
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        raise HTTPException(401, "Missing authentication token")
+    token = auth_header.split("Bearer ", 1)[1]
+    try:
+        decoded = fb_auth.verify_id_token(token)
+    except Exception:
+        raise HTTPException(401, "Invalid token")
+    
+    uid = decoded["uid"]
+    db = _init_firebase()
+    user_ref = db.collection("users").document(uid)
+    user_doc = user_ref.get()
+    
+    data = {
+        "email": decoded.get("email", ""),
+        "display_name": body.get("display_name", ""),
+        "updated_at": firestore.SERVER_TIMESTAMP
+    }
+    device_id = str(body.get("device_id", "")).strip()[:128]
+    
+    if not user_doc.exists:
+        data["created_at"] = firestore.SERVER_TIMESTAMP
+        data["is_blocked"] = False
+        data["device_ids"] = [device_id] if device_id else []
+        user_ref.set(data)
+    else:
+        if device_id:
+            existing = user_doc.to_dict() or {}
+            existing_devices = list(existing.get("device_ids") or [])
+            license_key = existing.get("license_key")
+            if not license_key:
+                data["device_ids"] = [device_id]
+            elif device_id not in existing_devices:
+                license_doc = db.collection("licenses").document(license_key).get()
+                lic = license_doc.to_dict() if license_doc.exists else {}
+                max_devices = max(1, int(lic.get("max_devices") or 1))
+                if len(existing_devices) >= max_devices:
+                    raise HTTPException(403, "Maximum licensed devices reached")
+                data["device_ids"] = firestore.ArrayUnion([device_id])
+        user_ref.update(data)
+        
+    return {"status": "success", "uid": uid}
+
+@app.post("/activate_license")
+async def activate_license(request: Request, body: dict = Body(...)):
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        raise HTTPException(401, "Missing token")
+    token = auth_header.split("Bearer ", 1)[1]
+    try:
+        decoded = fb_auth.verify_id_token(token)
+    except Exception:
+        raise HTTPException(401, "Invalid token")
+    
+    uid = decoded["uid"]
+    license_key = body.get("license_key")
+    if not license_key:
+        raise HTTPException(400, "license_key is required")
+        
+    db = _init_firebase()
+    license_ref = db.collection("licenses").document(license_key)
+    license_doc = license_ref.get()
+    
+    if not license_doc.exists:
+        raise HTTPException(404, "License key not found")
+        
+    lic = license_doc.to_dict()
+    current_status = lic.get("status")
+    if current_status not in ("unused", "unclaimed", "active"):
+        raise HTTPException(400, f"License status is {current_status}")
+        
+    if lic.get("assigned_uid") and lic.get("assigned_uid") != uid:
+        raise HTTPException(400, "License is already assigned to another user")
+        
+    from datetime import datetime, timedelta, timezone
+    
+    duration_days = lic.get("duration_days", 30)
+    expiry_date = lic.get("expiry_date")
+    device_id = str(request.headers.get("X-Device-ID", "")).strip()[:128]
+    user_ref = db.collection("users").document(uid)
+    @firestore.transactional
+    def claim_license(transaction):
+        fresh_snapshot = license_ref.get(transaction=transaction)
+        if not fresh_snapshot.exists:
+            raise HTTPException(404, "License key not found")
+        fresh = fresh_snapshot.to_dict() or {}
+        fresh_status = fresh.get("status")
+        if fresh_status not in ("unused", "unclaimed", "active"):
+            raise HTTPException(400, f"License status is {fresh_status}")
+        assigned_uid = fresh.get("assigned_uid")
+        if assigned_uid and assigned_uid != uid:
+            raise HTTPException(400, "License is already assigned to another user")
+        claimed_expiry = fresh.get("expiry_date")
+        updates = {
+            "status": "active",
+            "assigned_uid": uid,
+            "assigned_email": decoded.get("email", ""),
+        }
+        if not claimed_expiry:
+            claimed_expiry = datetime.now(timezone.utc) + timedelta(
+                days=int(fresh.get("duration_days") or duration_days)
+            )
+            updates["activated_at"] = firestore.SERVER_TIMESTAMP
+            updates["expiry_date"] = claimed_expiry
+        else:
+            exp_dt = (
+                claimed_expiry
+                if getattr(claimed_expiry, "tzinfo", None)
+                else claimed_expiry.replace(tzinfo=timezone.utc)
+            )
+            if exp_dt < datetime.now(timezone.utc):
+                raise HTTPException(400, "License has expired")
+        fresh_user_snapshot = user_ref.get(transaction=transaction)
+        fresh_user = fresh_user_snapshot.to_dict() if fresh_user_snapshot.exists else {}
+        device_ids = list(fresh_user.get("device_ids") or [])
+        max_devices = max(1, int(fresh.get("max_devices") or 1))
+        if len(device_ids) > max_devices:
+            raise HTTPException(403, "Maximum licensed devices reached")
+        if device_id and device_id not in device_ids:
+            if len(device_ids) >= max_devices:
+                raise HTTPException(403, "Maximum licensed devices reached")
+            device_ids.append(device_id)
+        user_clone_setting = (
+            bool(fresh_user.get("voice_clone_enabled"))
+            if "voice_clone_enabled" in fresh_user
+            else bool(fresh.get("voice_clone", False))
+        )
+        transaction.update(license_ref, updates)
+        transaction.set(user_ref, {
+            "license_key": license_key,
+            "license_status": "active",
+            "license_expiry": claimed_expiry,
+            "voice_clone_enabled": user_clone_setting,
+            "device_ids": device_ids,
+            "updated_at": firestore.SERVER_TIMESTAMP,
+        }, merge=True)
+        return claimed_expiry, user_clone_setting, fresh, dict(fresh_user, device_ids=device_ids)
+
+    expiry_date, user_clone_setting, lic, user_data = claim_license(db.transaction())
+    
+    # User-specific pool: dedicated Spaces first, otherwise public auto pool.
+    space_pool = _license_space_pool(db, uid, user_data, license_key, lic)
+
     return {
-        "status": "VoiceCraft TTS Server OK",
-        "engines": available_engines(),
-        "clone_enabled": clone_engines_enabled(),
+        "status": "success",
+        "plan_name": lic.get("plan_name", "Standard"),
+        "expiry_date": str(expiry_date) if expiry_date else None,
+        "voice_clone_enabled": user_clone_setting and bool(lic.get("voice_clone", False)),
+        "app_link": lic.get("app_link") or lic.get("appLink", ""),
+        "space_pool": space_pool
     }
 
+@app.get("/license_status")
+async def license_status(request: Request):
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        raise HTTPException(401, "Missing token")
+    token = auth_header.split("Bearer ", 1)[1]
+    try:
+        decoded = fb_auth.verify_id_token(token)
+    except Exception:
+        raise HTTPException(401, "Invalid token")
+        
+    uid = decoded["uid"]
+    db = _init_firebase()
+    user_doc = db.collection("users").document(uid).get()
+    
+    if not user_doc.exists:
+        return {"has_license": False, "status": "none"}
+        
+    user_data = user_doc.to_dict()
+    if user_data.get("is_blocked"):
+        return {"has_license": False, "status": "blocked"}
+
+    license_key = user_data.get("license_key")
+    if not license_key:
+        return {"has_license": False, "status": "none"}
+        
+    license_doc = db.collection("licenses").document(license_key).get()
+    if not license_doc.exists:
+        return {"has_license": False, "status": "none"}
+        
+    lic = license_doc.to_dict()
+    lic_status = lic.get("status", "active")
+    if lic_status != "active":
+        return {"has_license": False, "status": lic_status}
+
+    from datetime import datetime, timezone
+    expiry_date = lic.get("expiry_date")
+    is_expired = False
+    days_remaining = 30
+    if expiry_date:
+        try:
+            exp_dt = expiry_date if getattr(expiry_date, "tzinfo", None) else expiry_date.replace(tzinfo=timezone.utc)
+            delta = exp_dt - datetime.now(timezone.utc)
+            days_remaining = delta.days
+            if delta.total_seconds() < 0:
+                is_expired = True
+        except Exception:
+            pass
+
+    if is_expired:
+        return {"has_license": False, "status": "expired", "is_expired": True}
+
+    # User-specific pool: dedicated Spaces first, otherwise public auto pool.
+    space_pool = _license_space_pool(db, uid, user_data, license_key, lic)
 
-@app.get("/health")
-@app.head("/health")
-def health():
     return {
-        "status": "ok",
-        "piper_ready": PIPER_READY,
-        "silero_ready": SILERO_READY,
-        "silero_speakers": SILERO_SPEAKERS_RU,
-        "engines": available_engines(),
-        "clone_enabled": clone_engines_enabled(),
-        "zerogpu_enabled": False,
-        "clone_backend": CLONE_BACKEND_NAME,
-        "auth_mode": AUTH_MODE,
-        "api_auth_required": AUTH_MODE in {"api_secret", "layered"},
-        "license_auth_required": AUTH_MODE in {"license", "layered"},
-        "clone_model_ready": _pocket_clone_ready(),
-        "clone_auth_configured": _hf_token_configured(),
-        "clone_setup_required": clone_engines_enabled() and not _pocket_clone_ready(),
-        "clone_reference_cache_entries": len(_POCKET_VOICE_STATE_CACHE),
+        "has_license": True,
+        "status": "active",
+        "license_key": license_key,
+        "plan_name": lic.get("plan_name", "Pro"),
+        "expiry_date": str(expiry_date) if expiry_date else None,
+        "days_remaining": max(0, days_remaining),
+        "voice_clone_enabled": bool(user_data.get("voice_clone_enabled", lic.get("voice_clone", False))) and bool(lic.get("voice_clone", False)),
+        "app_link": lic.get("app_link") or lic.get("appLink", ""),
+        "space_pool": space_pool,
+        "is_expired": False
     }
 
+@app.get("/status")
+def status():
+    return {"status": "ok"}
+
+
+@app.get("/health")
+@app.head("/health")
+def health():
+    with _SPACE_STATE_LOCK:
+        active_jobs = _ACTIVE_JOBS
+    policy = get_runtime_policy()
+    clone_status = clone_backend_status()
+    firebase_configured = bool(HAS_FIREBASE and _init_firebase())
+    return {
+        "status": "ok",
+        "space_id": SPACE_ID,
+        "firebase_configured": firebase_configured,
+        "active_jobs": active_jobs,
+        "max_concurrent_jobs": MAX_CONCURRENT_JOBS,
+        "available": active_jobs < MAX_CONCURRENT_JOBS and bool(policy.get("space_enabled", True)) and not bool(policy.get("maintenance_mode", False)),
+        "clone_enabled": bool(policy.get("space_clone_enabled", False)),
+        "clone_ready": bool(clone_status.get("ready", False)),
+        "clone_status": clone_status.get("message", "unknown"),
+        "engines": BASE_ENGINES + (CLONE_ENGINES if policy.get("space_clone_enabled", False) else []),
+        "maintenance_mode": bool(policy.get("maintenance_mode", False)),
+        "version": SERVICE_VERSION,
+    }
+
 
 @app.get("/all_voices")
-async def all_voices_list():
-    """All voices - Edge, Piper, Silero, Anime styles, and CPU voice cloning."""
+async def all_voices_list(request: Request):
+    """All configured Edge, Piper, Silero, and CPU voice-cloning voices."""
+    await authorize_request(request, "edge")
     result = {"edge": {}, "piper": {}, "silero": {}}
 
-    # Edge TTS — 400+ voices (complete list, clean naming)
+    # Current Edge catalog. Desktop defaults to a smaller Featured view.
     try:
         import edge_tts
         voices = await edge_tts.list_voices()
@@ -1138,40 +1926,26 @@ async def all_voices_list():
     except Exception as e:
         print(f"Edge voices error: {e}")
 
-    # Piper TTS — all voices (clean naming, no engine hints)
-    try:
-        import urllib.request, json as _json
-        api_url = "https://huggingface.co/api/models/rhasspy/piper-voices"
-        req = urllib.request.Request(api_url, headers={"User-Agent": "VoiceCraft/2.0"})
-        with urllib.request.urlopen(req, timeout=60) as resp:
-            data = _json.loads(resp.read())
-        siblings = data.get("siblings", [])
-        for s in siblings:
-            rfn = s.get("rfilename", s.get("rfn", ""))
-            if not rfn.endswith(".onnx") or ".json" in rfn or "samples" in rfn:
-                continue
-            parts = rfn.split("/")
-            if len(parts) < 2:
-                continue
-            model_name = parts[-1].replace(".onnx", "")
-            dash_parts = model_name.rsplit("-", 2)
-            if len(dash_parts) >= 3:
-                lang_code = dash_parts[0].replace("_", "-")
-                voice = dash_parts[1].replace("_", " ").title()
-                quality = dash_parts[2]
-                if quality in ("low", "x_low"):
-                    continue
-                quality_map = {"high": " +", "medium": "", "low": " -", "x_low": " --"}
-                qs = quality_map.get(quality, " -")
-                display = f"{voice} [{lang_code}]{qs}"
-            else:
-                display = model_name.replace("_", " ").title()
-            full_code = f"piper:{model_name}"
-            result["piper"][display] = full_code
-    except Exception as e:
-        print(f"Piper dynamic fetch error: {e}")
-        for key in PIPER_VOICES:
-            result["piper"][key] = key  # voice code as string, not tuple
+    # Curated Piper catalog only; arbitrary repository models are not exposed.
+    piper_display_names = {
+        "piper:en_US-amy-medium": "Amy [en-US]",
+        "piper:en_US-joe-medium": "Joe [en-US]",
+        "piper:en_US-lessac-medium": "Lessac HQ [en-US]",
+        "piper:en_US-ryan-high": "Ryan HQ [en-US]",
+        "piper:en_GB-alan-medium": "Alan [en-GB]",
+        "piper:en_GB-alba-medium": "Alba [en-GB]",
+        "piper:ur_PK-fasih-medium": "Fasih [ur-PK]",
+        "piper:ar_JO-kareem-medium": "Kareem [ar-JO]",
+        "piper:hi_IN-pratham-medium": "Pratham [hi-IN]",
+        "piper:de_DE-thorsten-medium": "Thorsten [de-DE]",
+        "piper:ru_RU-irina-medium": "Irina [ru-RU]",
+        "piper:fr_FR-upmc-medium": "UPMC HQ [fr-FR]",
+        "piper:pt_BR-faber-medium": "Faber [pt-BR]",
+        "piper:tr_TR-dfki-medium": "Dfki [tr-TR]",
+        "piper:nl_NL-mls-medium": "MLS [nl-NL]",
+    }
+    for code in PIPER_VOICES:
+        result["piper"][piper_display_names[code]] = code
 
     # Silero v4 — Russian (official v4_ru speakers)
     silero_ru_speakers = {
@@ -1181,17 +1955,9 @@ async def all_voices_list():
     for speaker_code, display_name in silero_ru_speakers.items():
         result["silero"][f"{display_name} \u2022 Russian [RU]"] = f"silero:ru_{speaker_code}"
 
-    # Anime styles are lightweight Edge-based presets and remain CPU-safe.
-    anime_voices = {
-        "Anime Whisper [en]": "anime:en_whisper",
-        "Anime Edge [ja]": "anime:ja_edge",
-        "Anime Piper [zh]": "anime:zh_piper",
-    }
-    result["anime"] = anime_voices
-
-    if not clone_engines_enabled():
-        total = sum(len(group) for group in result.values())
-        return {"voices": result, "total": total, "clone_enabled": False}
+    if not clone_backend_available():
+        total = sum(len(group) for group in result.values())
+        return {"voices": result, "total": total, "clone_enabled": False}
 
     result["f5tts"] = {"Voice Clone": "f5tts:v1_base"}
     total = sum(len(group) for group in result.values())
@@ -1202,7 +1968,6 @@ GRADIO_DEFAULT_VOICES = {
     "edge": "en-US-AvaNeural",
     "piper": "piper:en_US-amy-medium",
     "silero": "silero:ru_xenia",
-    "anime": "anime:en_whisper",
     "f5tts": "f5tts:v1_base",
 }
 
@@ -1282,10 +2047,6 @@ def _build_gradio_ui(gr):
                 audio = await asyncio.to_thread(synthesize_silero, clean_text, voice)
                 return _write_audio_file(audio, ".wav"), "Ready - Silero audio generated."
 
-            if engine == "anime":
-                audio = await synthesize_anime(clean_text, voice, style)
-                return _write_audio_file(audio, ".mp3"), "Ready - anime-style audio generated."
-
             if engine == "f5tts":
                 audio = await synthesize_f5tts(clean_text, reference_path=reference_path)
                 return _write_audio_file(audio, ".wav"), "Ready - voice clone generated."
@@ -1349,14 +2110,7 @@ def _build_gradio_ui(gr):
 
 @app.get("/")
 def root():
-    return {
-        "name": "VoiceCraft TTS Server",
-        "status": "running",
-        "clone_enabled": clone_engines_enabled(),
-        "zerogpu_enabled": False,
-        "clone_backend": CLONE_BACKEND_NAME,
-        "endpoints": ["/tts", "/health", "/status", "/all_voices"],
-    }
+    return {"name": "VoiceCraft Service", "status": "ok"}
 
 
 if __name__ == "__main__":
@@ -1365,5 +2119,5 @@ if __name__ == "__main__":
         server_name="0.0.0.0",
         server_port=port,
         share=False,
-        show_error=True,
+        show_error=False,
     )

requirements-docker.txt

@@ -8,6 +8,7 @@ numpy>=2.0,<3
 scipy>=1.15.3
 soundfile>=0.12,<1
 pocket-tts>=2.1.0
+firebase-admin>=6.0.0
 
 # Docker profile is now CPU clone first. Silero v4 may be unavailable because
 # Pocket TTS requires a newer PyTorch runtime.

requirements.txt

@@ -8,3 +8,5 @@ numpy>=2.0,<3
 scipy>=1.15.3
 soundfile>=0.12,<1
 pocket-tts>=2.1.0
+firebase-admin>=6.0.0
+