diff --git a/app/ai/voice/tts/dragontts.py b/app/ai/voice/tts/dragontts.py index 2953156de..06744c311 100644 --- a/app/ai/voice/tts/dragontts.py +++ b/app/ai/voice/tts/dragontts.py @@ -23,6 +23,7 @@ from __future__ import annotations +import json from collections.abc import AsyncGenerator from dataclasses import dataclass, field from typing import TYPE_CHECKING, Optional @@ -242,7 +243,28 @@ async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, Non async with self._client.stream( "POST", f"{self._url}/tts/stream", json=body ) as response: - response.raise_for_status() + if response.status_code >= 400: + # Read the error body WHILE the stream context is open — + # after it closes a streaming response's body is gone, so + # raise_for_status()+e.response.text loses the reason. + # DragonTTS returns {"detail": "upstream returned + # an error: "} on a provider failure; surface THAT + # so the real cause (e.g. a Gemini error) reaches the call, + # not a generic "DragonTTS stream failed". + err = await response.aread() + try: + detail = json.loads(err).get( + "detail", err.decode("utf-8", "replace") + ) + except Exception: + detail = err.decode("utf-8", "replace") + msg = ( + f"DragonTTS /tts/stream HTTP {response.status_code}: " + f"{str(detail)[:300]}" + ) + logger.error(msg) + yield ErrorFrame(error=msg) + return async for chunk in response.aiter_bytes(): if not chunk: continue @@ -260,12 +282,6 @@ async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, Non ) # A trailing odd byte (carry is always 0 or 1 byte here) can't form a # complete 16-bit sample, so drop it rather than emit a corrupt frame. - except httpx.HTTPStatusError as e: - detail = e.response.text[:200] - logger.error(f"DragonTTS returned HTTP {e.response.status_code}: {detail}") - yield ErrorFrame( - error=f"DragonTTS returned HTTP {e.response.status_code}: {detail}" - ) except httpx.RequestError as e: logger.error(f"DragonTTS stream request failed: {e}") yield ErrorFrame(error=f"DragonTTS stream request failed: {e}") diff --git a/app/api/routers/breeze_buddy/dragontts.py b/app/api/routers/breeze_buddy/dragontts.py index 5a9de4e7d..4107c85b3 100644 --- a/app/api/routers/breeze_buddy/dragontts.py +++ b/app/api/routers/breeze_buddy/dragontts.py @@ -11,13 +11,15 @@ unhealthy (1->0); the manage endpoint is the only way to restore it. """ -from fastapi import APIRouter, Depends +import httpx +from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response from app.ai.voice.agents.breeze_buddy.tts.dragontts.kill_switch import ( get_dragontts_status, set_dragontts_health, ) from app.api.security.breeze_buddy.rbac_token import get_current_user_with_rbac +from app.core.config.dynamic import DRAGONTTS_URL from app.core.security.authorization import require_admin from app.schemas import UserInfo from app.schemas.breeze_buddy.dragontts import ( @@ -53,3 +55,107 @@ async def get_dragon_tts_status( """Return the current DragonTTS health-gate state (health + caching).""" require_admin(current_user) return DragonTTSStatus(**await get_dragontts_status()) + + +# --- DragonTTS cache browser/management proxy (admin-only) ------------------- +# DragonTTS is NOT internet-exposed; the dashboard reaches its cache API through +# these admin routes, which forward to the internal service and relay the +# response verbatim (JSON or binary audio). Auth matches the manage/status +# routes above: get_current_user_with_rbac (Depends) + in-body require_admin. + +_PROXY_TIMEOUT = 30.0 + + +async def _forward(method: str, path: str, **kwargs) -> Response: + """Forward a request to the internal DragonTTS service and relay the + response (status, bytes, content-type). HTTPException(502) if DragonTTS is + unreachable; otherwise DragonTTS's own status is mirrored (e.g. 404).""" + url = f"{await DRAGONTTS_URL()}{path}" + try: + async with httpx.AsyncClient(timeout=_PROXY_TIMEOUT) as client: + resp = await client.request(method, url, **kwargs) + except httpx.RequestError as e: + raise HTTPException(status_code=502, detail=f"DragonTTS unreachable: {e}") + # Relay DragonTTS's response verbatim, INCLUDING non-2xx. DragonTTS already + # returns a JSON {"detail": "upstream returned an error: ..."} + # with the real provider reason; re-raising as HTTPException(detail=resp.text) + # would double-wrap it ({detail: "{\"detail\": ...}"}) and swap the + # content-type, so the dashboard would lose the structured body. Mirror the + # upstream status + body + content-type as-is. + return Response( + content=resp.content, + media_type=resp.headers.get("content-type"), + status_code=resp.status_code, + ) + + +@router.get("/admin/dragontts/cache/by-text") +async def proxy_cache_by_text( + text: str = Query(..., description="text to look up"), + match: str = Query("exact", description="'exact' or 'substring'"), + provider: str | None = None, + voice_id: str | None = None, + include_audio: bool = Query(False, description="inline base64 audio per entry"), + limit: int = Query(100, ge=1, le=1000), + current_user: UserInfo = Depends(get_current_user_with_rbac), +): + """Browse every cached variant of a text (metadata + params + hash, optional + inline audio). Proxied to DragonTTS GET /cache/by-text.""" + require_admin(current_user) + params = { + "text": text, + "match": match, + "include_audio": include_audio, + "limit": limit, + } + if provider is not None: + params["provider"] = provider + if voice_id is not None: + params["voice_id"] = voice_id + return await _forward("GET", "/cache/by-text", params=params) + + +@router.get("/admin/dragontts/cache/{key}") +async def proxy_cache_get( + key: str, current_user: UserInfo = Depends(get_current_user_with_rbac) +): + """Download a cached entry's audio by hash. Proxied to DragonTTS GET /cache/{key}.""" + require_admin(current_user) + return await _forward("GET", f"/cache/{key}") + + +@router.delete("/admin/dragontts/cache/{key}") +async def proxy_cache_delete( + key: str, current_user: UserInfo = Depends(get_current_user_with_rbac) +): + """Delete a cached entry by hash. Proxied to DragonTTS DELETE /cache/{key}.""" + require_admin(current_user) + return await _forward("DELETE", f"/cache/{key}") + + +@router.post("/admin/dragontts/cache/{key}/resynth") +async def proxy_cache_resynth( + key: str, current_user: UserInfo = Depends(get_current_user_with_rbac) +): + """Re-synthesize an entry from its stored metadata. Proxied to DragonTTS + POST /cache/{key}/resynth.""" + require_admin(current_user) + return await _forward("POST", f"/cache/{key}/resynth") + + +@router.post("/admin/dragontts/cache/{key}/audio") +async def proxy_cache_replace_audio( + key: str, + request: Request, + current_user: UserInfo = Depends(get_current_user_with_rbac), +): + """Replace an entry's audio with a user-supplied WAV (base64 JSON body). + Proxied verbatim to DragonTTS POST /cache/{key}/audio.""" + require_admin(current_user) + raw = await request.body() + return await _forward( + "POST", + f"/cache/{key}/audio", + content=raw, + headers={"content-type": "application/json"}, + ) diff --git a/dragontts/app/api/v1/cache.py b/dragontts/app/api/v1/cache.py index 5b2ec9ea6..0a04f7daf 100644 --- a/dragontts/app/api/v1/cache.py +++ b/dragontts/app/api/v1/cache.py @@ -2,17 +2,82 @@ from __future__ import annotations -from datetime import datetime +import base64 +import json +from datetime import datetime, timedelta, timezone from fastapi import APIRouter, HTTPException, Query, Request, Response -from pydantic import BaseModel +from pydantic import BaseModel, Field from app.alerts.summary import send_daily_summary from app.audio.format import content_type_for -from app.schemas.cache import CacheEntryInfo, PaginatedCache +from app.core.config import settings +from app.core.logging import logger +from app.schemas.cache import ( + ByTextResponse, + CacheEntryDetail, + CacheEntryInfo, + PaginatedCache, +) router = APIRouter() +# Bound analytics date ranges so a wide query can't CPU-bomb the single pod. +# None/None -> last N days (NOT all-time, which would full-scan the tables). +ANALYTICS_MAX_RANGE_DAYS = settings.analytics_max_range_days +# Cap entries returned WITH inline audio (each carries a base64 blob), so +# include_audio=true can't build a multi-hundred-MB JSON response. +BY_TEXT_AUDIO_LIMIT = 25 +# Cap a user-supplied audio upload (base64 of a WAV recording). +MAX_AUDIO_UPLOAD_B64_BYTES = 12_000_000 # ~9 MB decoded WAV +# Supported text-match modes. Anything else used to silently fall through to +# exact matching; reject it so a typo surfaces as 400 instead of wrong results. +ALLOWED_MATCH = ("exact", "substring") + + +def _validate_match(match: str) -> str: + """Reject unknown match modes (only exact/substring are supported).""" + if match not in ALLOWED_MATCH: + raise HTTPException( + status_code=400, detail=f"match must be one of {ALLOWED_MATCH}" + ) + return match + + +def _bound_range(from_date: str | None, to_date: str | None) -> tuple[str, str]: + """Parse + bound an analytics date range to <= ANALYTICS_MAX_RANGE_DAYS. + + Returns (from_iso, to_iso). Unset bounds default to the last N days (not + all-time, which would full-scan the tables). Raises HTTPException(400) on + bad format, inverted range, or a range wider than the cap. + """ + today = datetime.now(timezone.utc).date() + + def _parse(s: str | None): + if s is None: + return None + try: + return datetime.strptime(s, "%Y-%m-%d").date() + except ValueError: + raise HTTPException( + status_code=400, detail=f"invalid date {s!r}; use YYYY-MM-DD" + ) + + f = _parse(from_date) + t = _parse(to_date) + n = ANALYTICS_MAX_RANGE_DAYS + if f is None and t is None: + f, t = today - timedelta(days=n - 1), today + elif f is None: + f = t - timedelta(days=n - 1) + elif t is None: + t = min(f + timedelta(days=n - 1), today) + if t < f: + raise HTTPException(status_code=400, detail="'to' must be on/after 'from'") + if (t - f).days + 1 > n: + raise HTTPException(status_code=400, detail=f"date range exceeds {n} days") + return f.isoformat(), t.isoformat() + def _to_info(r) -> CacheEntryInfo: return CacheEntryInfo( @@ -31,6 +96,31 @@ def _to_info(r) -> CacheEntryInfo: ) +def _to_detail(r, audio_b64: str | None = None) -> CacheEntryDetail: + try: + params = json.loads(r.params) if r.params else {} + except (ValueError, TypeError): + params = {} + return CacheEntryDetail( + key=r.key, + text=r.text, + provider=r.provider, + voice_id=r.voice_id, + model=r.model, + language=r.language, + params=params, + container=r.container, + encoding=r.encoding, + sample_rate=r.sample_rate, + size_bytes=r.size_bytes, + hit_count=r.hit_count, + created_at=r.created_at, + last_accessed_at=r.last_accessed_at, + ttl_expires_at=r.ttl_expires_at, + audio_base64=audio_b64, + ) + + MAX_CACHE_LIST_LIMIT = 1000 # hard cap so a list call can never stream the whole DB @@ -53,6 +143,7 @@ async def list_cache( """Paginated entry listing with optional text/date filters. limit is clamped to [1, MAX_CACHE_LIST_LIMIT]. q+match filters text; created_after is inclusive, created_before is exclusive (both YYYY-MM-DD).""" + _validate_match(match) metadata = request.app.state.metadata limit = max(1, min(limit, MAX_CACHE_LIST_LIMIT)) offset = max(0, offset) @@ -83,14 +174,7 @@ async def stats( """Cache metrics. Request metrics are date-filterable (?from=&to= YYYY-MM-DD); the cache snapshot is the current point-in-time state. All from stored rollups/totals — no full-table scan per call.""" - for label, value in (("from", from_date), ("to", to_date)): - if value is not None: - try: - datetime.strptime(value, "%Y-%m-%d") - except ValueError: - raise HTTPException( - status_code=400, detail=f"invalid {label}; use YYYY-MM-DD" - ) + from_date, to_date = _bound_range(from_date, to_date) metadata = request.app.state.metadata registry = request.app.state.registry metrics = await metadata.metrics_summary(from_date=from_date, to_date=to_date) @@ -124,14 +208,7 @@ async def stats_daily( words_from_cache_pct derived) plus a per-provider breakdown. ``provider`` narrows the view to one provider (its totals then reflect that provider only). ?from=&to= YYYY-MM-DD (UTC dates).""" - for label, value in (("from", from_date), ("to", to_date)): - if value is not None: - try: - datetime.strptime(value, "%Y-%m-%d") - except ValueError: - raise HTTPException( - status_code=400, detail=f"invalid {label}; use YYYY-MM-DD" - ) + from_date, to_date = _bound_range(from_date, to_date) cache = request.app.state.cache return await cache.daily_stats( from_date=from_date, to_date=to_date, provider=provider @@ -150,14 +227,7 @@ async def stats_latency( ``cache_serve`` (HIT serve), ``ttfb`` (stream first byte), plus a derived ``miss_overhead_us`` = total.avg - synth.avg (DragonTTS work beyond the provider call). ?from=&to= YYYY-MM-DD (UTC); ?provider= narrows to one.""" - for label, value in (("from", from_date), ("to", to_date)): - if value is not None: - try: - datetime.strptime(value, "%Y-%m-%d") - except ValueError: - raise HTTPException( - status_code=400, detail=f"invalid {label}; use YYYY-MM-DD" - ) + from_date, to_date = _bound_range(from_date, to_date) metadata = request.app.state.metadata providers = await metadata.latency_summary_by_provider( from_date=from_date, to_date=to_date, provider=provider @@ -219,6 +289,7 @@ async def delete_by_text(body: DeleteByTextBody, request: Request): provider/voice_id. ``dry_run`` defaults True — previews matched entries + keys without deleting; set ``dry_run: false`` to delete. Requires at least one filter (text/provider/voice_id) — use /cache/clear to wipe everything.""" + _validate_match(body.match) cache = request.app.state.cache try: return await cache.delete_by_text( @@ -265,6 +336,119 @@ async def slack_summary(request: Request): return {"sent": sent} +@router.get("/cache/by-text", response_model=ByTextResponse) +async def cache_by_text( + request: Request, + text: str = Query(..., description="text to look up"), + match: str = Query("exact", description="'exact' (default) or 'substring'"), + provider: str | None = None, + voice_id: str | None = None, + include_audio: bool = Query(False, description="base64-encode each clip's audio"), + limit: int = Query(100, ge=1, le=MAX_CACHE_LIST_LIMIT), +): + """Look up every cached entry for a text (all provider/model/voice/param + variants). Returns full metadata + parsed params + hash; optionally inline + audio (base64) when include_audio=true.""" + _validate_match(match) + metadata = request.app.state.metadata + blobs = request.app.state.blobs + if include_audio: + limit = min(limit, BY_TEXT_AUDIO_LIMIT) + records = await metadata.list( + provider=provider, + voice_id=voice_id, + limit=limit, + offset=0, + q=text, + match=match, + ) + entries = [] + for r in records: + audio_b64 = None + if include_audio: + try: + audio_b64 = base64.b64encode(await blobs.get(r.storage_path)).decode( + "ascii" + ) + except Exception as e: # missing blob — skip audio, keep metadata + logger.warning(f"blob read failed for {r.key}: {e}") + entries.append(_to_detail(r, audio_b64)) + return ByTextResponse(entries=entries, count=len(entries)) + + +@router.post("/cache/{key}/resynth") +async def resynth_cache(key: str, request: Request): + """Re-synthesize an entry from its stored metadata (same key, fresh audio).""" + cache = request.app.state.cache + try: + key, status, source, size, provider, model, enc, rate = ( + await cache.resynth_by_key(key) + ) + except KeyError: + raise HTTPException(status_code=404, detail="cache key not found") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except ( + Exception + ) as e: # provider synth/upload failure (ProviderError/httpx) -> 502, not 500 + raise HTTPException(status_code=502, detail=f"upstream failure: {e}") + return { + "key": key, + "status": status, + "source": source, + "size_bytes": size, + "provider": provider, + "model": model, + "encoding": enc, + "sample_rate": rate, + } + + +class ReplaceAudioRequest(BaseModel): + """Body for POST /cache/{key}/audio — a user-supplied WAV recording as + base64. Mirrors the ``audio_base64`` convention of /tts/create (no multipart + dependency).""" + + audio_base64: str = Field( + ..., + max_length=MAX_AUDIO_UPLOAD_B64_BYTES, + description="WAV recording, base64-encoded (<=~9MB)", + ) + + +@router.post("/cache/{key}/audio") +async def replace_cache_audio(key: str, body: ReplaceAudioRequest, request: Request): + """Replace an entry's audio blob with a user-supplied WAV recording (base64). + The upload is decoded + converted to the entry's native format.""" + cache = request.app.state.cache + try: + data = base64.b64decode(body.audio_base64) + except Exception as e: + raise HTTPException(status_code=400, detail=f"invalid audio_base64: {e}") + try: + key, status, source, size, provider, model, enc, rate = ( + await cache.replace_audio_by_key(key, data) + ) + except KeyError: + raise HTTPException(status_code=404, detail="cache key not found") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except ( + Exception + ) as e: # provider synth/upload failure (ProviderError/httpx) -> 502, not 500 + raise HTTPException(status_code=502, detail=f"upstream failure: {e}") + return { + "key": key, + "status": status, + "source": source, + "size_bytes": size, + "provider": provider, + "model": model, + "encoding": enc, + "sample_rate": rate, + } + + @router.get("/cache/{key}") async def get_cache(key: str, request: Request): metadata = request.app.state.metadata diff --git a/dragontts/app/api/v1/tts.py b/dragontts/app/api/v1/tts.py index baa364e72..bb6aa6a24 100644 --- a/dragontts/app/api/v1/tts.py +++ b/dragontts/app/api/v1/tts.py @@ -63,6 +63,11 @@ async def tts_bytes(req: TTSRequest, request: Request): ) except (httpx.HTTPStatusError, httpx.RequestError) as e: raise _map_upstream_error(provider, e) + except Exception as e: # any other provider/lib error -> 502 with the reason + raise HTTPException( + status_code=502, + detail=f"upstream {provider} error: {type(e).__name__}: {e}", + ) return Response( content=audio, @@ -101,6 +106,11 @@ async def tts_stream(req: TTSRequest, request: Request): raise HTTPException( status_code=502, detail=f"upstream {provider} returned an error: {e}" ) + except Exception as e: # any other provider/lib error -> 502 with the reason + raise HTTPException( + status_code=502, + detail=f"upstream {provider} error: {type(e).__name__}: {e}", + ) try: first = await gen.__anext__() @@ -114,6 +124,12 @@ async def tts_stream(req: TTSRequest, request: Request): except (ProviderError, httpx.HTTPStatusError, httpx.RequestError, OSError) as e: await gen.aclose() raise _map_upstream_error(provider, e) + except Exception as e: # any other provider/lib error -> 502 with the reason + await gen.aclose() + raise HTTPException( + status_code=502, + detail=f"upstream {provider} error: {type(e).__name__}: {e}", + ) async def body(first_chunk: bytes) -> AsyncGenerator[bytes, None]: try: @@ -186,6 +202,11 @@ async def tts_create(req: TTSRequest, request: Request): ) except (httpx.HTTPStatusError, httpx.RequestError) as e: raise _map_upstream_error(provider, e) + except Exception as e: # any other provider/lib error -> 502 with the reason + raise HTTPException( + status_code=502, + detail=f"upstream {provider} error: {type(e).__name__}: {e}", + ) return CreateResponse( key=key, diff --git a/dragontts/app/cache/service.py b/dragontts/app/cache/service.py index ad0a53b38..d8de1fd78 100644 --- a/dragontts/app/cache/service.py +++ b/dragontts/app/cache/service.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio +import json import random import re import time @@ -37,7 +38,7 @@ from app.core.logging import logger from app.providers.base import AudioResult, BaseTTSProvider, ProviderError from app.providers.registry import ProviderNotConfigured -from app.schemas.tts import TTSRequest +from app.schemas.tts import CartesiaVoice, OutputFormat, TTSRequest from app.storage.base import CacheRecord, escape_like @@ -46,6 +47,50 @@ def _wc(text: str | None) -> int: return len((text or "").split()) +def _decode_upload_to_pcm(data: bytes) -> tuple[bytes, int]: + """Decode an uploaded clip to mono PCM s16le + its sample rate. + + Accepts WAV (parsed via stdlib ``wave``); raises ``ValueError`` (which the + API layer maps to HTTP 400) on anything else — INCLUDING malformed frames + that make ``audioop`` raise (``audioop.error`` is NOT a ``ValueError``, so + the tomono/lin2lin calls are inside the try too). MP3/ogg are NOT supported + (no ffmpeg/pydub). Stereo is down-mixed and non-16-bit widened. Duration is + capped to bound memory. + """ + import audioop + import wave + from io import BytesIO + + MAX_DURATION_S = 120 # reject absurdly long uploads before allocating PCM + try: + with wave.open(BytesIO(data), "rb") as wf: + nchannels = wf.getnchannels() + sampwidth = wf.getsampwidth() + framerate = wf.getframerate() + nframes = wf.getnframes() + if ( + framerate <= 0 + or framerate > 96000 + or sampwidth not in (1, 2, 3, 4) + or nchannels not in (1, 2) + or nframes / framerate > MAX_DURATION_S + ): + raise ValueError( + f"unsupported WAV: {nchannels}ch {sampwidth}*{framerate}Hz " + f"{nframes} frames" + ) + frames = wf.readframes(nframes) + if nchannels > 1: + frames = audioop.tomono(frames, sampwidth, 1.0, 1.0) + if sampwidth != 2: + frames = audioop.lin2lin(frames, sampwidth, 2) + except ValueError: + raise + except Exception as e: # malformed/not-WAV, OR audioop.error on odd frames + raise ValueError(f"upload is not a decodable WAV file: {e}") from e + return frames, framerate + + # Sentinel: caller has NOT pre-fetched the existing record (vs. None = checked, # genuinely absent). Avoids a redundant get() on every store when the caller # already knows. @@ -1096,6 +1141,78 @@ async def create(self, req: TTSRequest, audio_override: bytes | None = None): store_rate, ) + def _request_from_record(self, record: CacheRecord) -> TTSRequest: + """Rebuild a TTSRequest from a stored CacheRecord. + + Used by resynth_by_key / replace_audio_by_key so the derived key + round-trips (the same provider/model/text/language/params/output_format + that originally produced ``record.key`` must re-derive to it). + """ + try: + params = json.loads(record.params) if record.params else {} + except (ValueError, TypeError): + params = {} + return TTSRequest( + model_id=f"{record.provider}:{record.model}", + transcript=record.text or "", + voice=CartesiaVoice(id=record.voice_id), + language=record.language or "", + output_format=OutputFormat( + container=record.container, + encoding=record.encoding, + sample_rate=record.sample_rate, + ), + params=params, + ) + + async def resynth_by_key(self, key: str): + """Re-synthesize an entry from its stored metadata, replacing the blob. + + Reads the record, rebuilds the request, asserts the key round-trips, and + delegates to :meth:`create` (forces a fresh synth with ``replace=True``). + Raises ``KeyError`` if the key is absent; ``ValueError`` if the stored + fields no longer re-derive to the same key (the API maps that to 400). + Returns the same tuple as :meth:`create`. + """ + record = await self._metadata.get(key) + if not record: + raise KeyError(key) + req = self._request_from_record(record) + _p, _m, _of, _canon, derived = self._resolve(req) + if derived != key: + raise ValueError( + f"key round-trip mismatch: stored {key!r} vs derived {derived!r}" + ) + return await self.create(req) + + async def replace_audio_by_key(self, key: str, audio: bytes): + """Replace an entry's audio blob with a user-supplied recording. + + Decodes the upload (WAV) to mono PCM s16le, converts to the entry's + native format, and delegates to :meth:`create` with ``audio_override`` + (stores under the same key with ``replace=True``). Raises ``KeyError`` / + ``ValueError`` like :meth:`resynth_by_key`; ``ValueError`` also covers an + undecodable upload. Returns the same tuple as :meth:`create`. + """ + record = await self._metadata.get(key) + if not record: + raise KeyError(key) + native_pcm, native_rate = _decode_upload_to_pcm(audio) # raises ValueError + native = await self._convert_audio( + native_pcm, + native_encoding="pcm_s16le", + native_rate=native_rate, + out_encoding=record.encoding, + out_rate=record.sample_rate, + ) + req = self._request_from_record(record) + _p, _m, _of, _canon, derived = self._resolve(req) + if derived != key: + raise ValueError( + f"key round-trip mismatch: stored {key!r} vs derived {derived!r}" + ) + return await self.create(req, audio_override=native) + async def delete(self, req: TTSRequest): """Delete by derived key. Returns (deleted_bool, key). diff --git a/dragontts/app/core/config.py b/dragontts/app/core/config.py index 95a491da1..c3a05ad9d 100644 --- a/dragontts/app/core/config.py +++ b/dragontts/app/core/config.py @@ -208,6 +208,14 @@ def _coerce_residency_url(cls, v: str) -> str: # latency_samples rows older than this are pruned by the periodic checkpoint # loop, keeping the table bounded. metrics_latency_retention_days: int = 14 + # Max width (days) of any analytics date range (/stats, /stats/daily, + # /stats/latency). Bounds the aggregation so a wide "last year" range can't + # CPU-bomb the single pod; unset from/to defaults to the last N days (not + # all-time, which would full-scan metrics_daily/latency_samples). Raise for + # longer cache-hit trend windows — metrics_daily is 1 row/day, so even 90 + # days is cheap; the real cost is latency_samples p95 sorts, hence the tight + # default matching its 14-day retention. + analytics_max_range_days: int = 10 # --- Slack daily summary (mirrors clairvoyance's incoming-webhook pattern) --- # Off by default: an empty SLACK_WEBHOOK_URL disables the feature (no separate diff --git a/dragontts/app/providers/elevenlabs.py b/dragontts/app/providers/elevenlabs.py index 3709685ac..fbad82014 100644 --- a/dragontts/app/providers/elevenlabs.py +++ b/dragontts/app/providers/elevenlabs.py @@ -29,6 +29,11 @@ from app.providers import elevenlabs_pool from app.providers.base import AudioResult, BaseTTSProvider, ProviderError +# Models that accept a language_code (the multilingual flash/turbo variants). +# v3 and multilingual_v2 auto-detect language; English-only models ignore it. +# Mirrors pipecat's ELEVENLABS_MULTILINGUAL_MODELS. +_ELEVENLABS_MULTILINGUAL_MODELS = {"eleven_flash_v2_5", "eleven_turbo_v2_5"} + class ElevenLabsProvider(BaseTTSProvider): """ElevenLabs text-to-speech adapter. @@ -66,11 +71,15 @@ def __init__(self, api_key: str | None = None, base_url: str | None = None) -> N else settings.elevenlabs_indian_residency_base_url ) self._client = httpx.AsyncClient(timeout=30.0) - # One warm pool per (voice_id, model_id): the WS binds voice (URL path) - # and model_id (connect-time query param) for the socket's lifetime, so - # differing voice/model need separate sockets. Lazily created on first - # streaming miss; warmed eagerly when the model matches a default. - self._pools: dict[tuple[str, str], elevenlabs_pool.ElevenLabsStreamPool] = {} + # One warm pool per (voice_id, model_id, enable_ssml_parsing, language): + # the WS binds voice (URL path) and model_id (connect-time query param) + # for the socket's lifetime, and SSML + language are connect-time socket + # settings (see _get_pool), so differing values need separate sockets. + # Lazily created on first streaming miss; warmed eagerly when the model + # matches a default. + self._pools: dict[ + tuple[str, str, bool, str | None], elevenlabs_pool.ElevenLabsStreamPool + ] = {} def _voice_settings(self, params: dict) -> dict: # Caller-supplied voice_settings win; otherwise mirror the one-shot @@ -89,21 +98,32 @@ def _voice_settings(self, params: dict) -> dict: return {"stability": 0.5, "similarity_boost": 0.75, "speed": speed} def _get_pool( - self, voice_id: str, model_id: str, enable_ssml_parsing: bool = False + self, + voice_id: str, + model_id: str, + enable_ssml_parsing: bool = False, + language: str | None = None, ) -> elevenlabs_pool.ElevenLabsStreamPool | None: - """Return the warm pool for (voice, model, ssml), creating it lazily. + """Return the warm pool for (voice, model, ssml, language), creating it lazily. Returns ``None`` when pooling is disabled (pool size 0) or the key is missing, so the caller falls back to one-shot synth. NB: ``_pools`` grows - with distinct (voice, model, ssml) triples and is only cleared at - shutdown — acceptable while the voice catalog stays fixed (re-add an LRU - cap if it ever diversifies). SSML is part of the key because it's a - connect-time socket setting: an SSML-on socket parses tags for - ALL its utterances, so on/off can't share one socket. + with distinct (voice, model, ssml, language) tuples and is only cleared + at shutdown — acceptable while the voice catalog stays fixed (re-add an + LRU cap if it ever diversifies). SSML and language are part of the key + because they're connect-time socket settings: an SSML-on socket parses + tags for ALL its utterances, and language_code pins the socket's + language, so differing values can't share one socket. """ if not self.api_key or settings.elevenlabs_stream_pool_size < 1: return None - key = (voice_id, model_id, enable_ssml_parsing) + # Non-multilingual models ignore language on the socket (the pool only + # sends language_code for multilingual models), so normalize it out of + # the key — otherwise identical requests that differ only by language + # each spin up a redundant warm socket (pool fragmentation). + if model_id not in _ELEVENLABS_MULTILINGUAL_MODELS: + language = None + key = (voice_id, model_id, enable_ssml_parsing, language) pool = self._pools.get(key) if pool is None: pool = elevenlabs_pool.ElevenLabsStreamPool( @@ -118,6 +138,7 @@ def _get_pool( settings.elevenlabs_stream_pool_size + 4, ), enable_ssml_parsing=enable_ssml_parsing, + language=language, ) self._pools[key] = pool return pool @@ -132,10 +153,11 @@ async def warm(self) -> None: defaults = PROVIDER_DEFAULTS.get("elevenlabs", {}) voice = defaults.get("voice_id", "") model = defaults.get("model", "eleven_flash_v2_5") + language = defaults.get("language") if not voice or not model: return try: - pool = self._get_pool(voice, model) + pool = self._get_pool(voice, model, language=language) if pool is not None: await pool.start() except Exception as e: @@ -192,14 +214,16 @@ async def synth( "Content-Type": "application/json", "Accept": "audio/raw", } - # NOTE: language_code is intentionally NOT sent — the ElevenLabs - # /v1/text-to-speech/{voice_id} body has no such field (language is - # bound to the model), and clairvoyance's reference doesn't send it. payload = { "text": text, "model_id": final_model_id, "voice_settings": self._voice_settings(params), } + # language_code: only the multilingual models (flash_v2_5/turbo_v2_5) + # accept it; v3/multilingual_v2 auto-detect, English-only models ignore + # it. Send the base subtag ("en"/"hi") — matches pipecat's use_base_code. + if final_model_id in _ELEVENLABS_MULTILINGUAL_MODELS and final_language: + payload["language_code"] = final_language.split("-")[0] if params.get("enable_ssml_parsing"): # SSML on: ElevenLabs parses etc. into real # pauses instead of reading the tags aloud. Default off; only sent @@ -259,7 +283,8 @@ async def stream_synth( f"[voice_id={final_voice_id}, model_id={final_model_id}, ssml={ssml}]" ) - pool = self._get_pool(final_voice_id, final_model_id, ssml) + lang_code = final_language.split("-")[0] if final_language else None + pool = self._get_pool(final_voice_id, final_model_id, ssml, lang_code) if pool is not None: streamed_any = False try: diff --git a/dragontts/app/providers/elevenlabs_pool.py b/dragontts/app/providers/elevenlabs_pool.py index b02d3c4d5..e49b56a48 100644 --- a/dragontts/app/providers/elevenlabs_pool.py +++ b/dragontts/app/providers/elevenlabs_pool.py @@ -47,6 +47,10 @@ from app.core.logging import logger from app.providers.base import ProviderError +# Models that accept a language_code query param on the multi-context socket. +# Mirrors pipecat's ELEVENLABS_MULTILINGUAL_MODELS. +_ELEVENLABS_MULTILINGUAL_MODELS = {"eleven_flash_v2_5", "eleven_turbo_v2_5"} + # Path appended to the WS host. The host is derived from the provider's # base_url (https->wss) so the Indian-residency instance streams against # wss://api.in.residency.elevenlabs.io rather than the global endpoint. @@ -200,6 +204,9 @@ def __init__( # SSML-on and SSML-off need separate warm sockets (the provider keys the # pool by this flag). Appended to the connect URI as a query param. enable_ssml_parsing: bool = False, + # language_code is a connect-time query param (like model_id/ssml), so + # the pool is also keyed by it (see provider._get_pool). None = omit. + language: str | None = None, ): if not api_key: raise ValueError("ElevenLabsStreamPool requires an api_key") @@ -231,6 +238,10 @@ def __init__( # Connection-level: the socket parses SSML tags in every utterance # sent over it (matches pipecat's connect-URI query param). self._uri += "&enable_ssml_parsing=true" + if language and model_id in _ELEVENLABS_MULTILINGUAL_MODELS: + # Multilingual models (flash_v2_5/turbo_v2_5) take a language_code; + # others auto-detect or ignore it. Connect-time, so pool-keyed. + self._uri += f"&language_code={quote(language, safe='')}" self._headers = {"xi-api-key": api_key} self._connect_fn = connect_fn or _default_connect self._min_size = min_size diff --git a/dragontts/app/schemas/cache.py b/dragontts/app/schemas/cache.py index 444b93c03..54cac99ef 100644 --- a/dragontts/app/schemas/cache.py +++ b/dragontts/app/schemas/cache.py @@ -2,7 +2,7 @@ from __future__ import annotations -from pydantic import BaseModel +from pydantic import BaseModel, Field class CacheEntryInfo(BaseModel): @@ -59,3 +59,33 @@ class CreateResponse(BaseModel): class DeleteResponse(BaseModel): status: str # "deleted" | "not_found" key: str + + +class CacheEntryDetail(BaseModel): + """Full cache entry — metadata + parsed params + optional inline audio. + + Used by GET /cache/by-text (the browse/inspect endpoint). ``params`` is the + parsed canonical-JSON tuning map; ``audio_base64`` is present only when the + caller passes ``include_audio=true``.""" + + key: str + text: str | None = None + provider: str + voice_id: str + model: str + language: str + params: dict = Field(default_factory=dict) + container: str + encoding: str + sample_rate: int + size_bytes: int + hit_count: int + created_at: str + last_accessed_at: str | None = None + ttl_expires_at: str | None = None + audio_base64: str | None = None + + +class ByTextResponse(BaseModel): + entries: list[CacheEntryDetail] + count: int diff --git a/dragontts/app/storage/sqlite.py b/dragontts/app/storage/sqlite.py index 38ad5ec0d..d7636840d 100644 --- a/dragontts/app/storage/sqlite.py +++ b/dragontts/app/storage/sqlite.py @@ -42,6 +42,10 @@ ttl_expires_at TEXT ); CREATE INDEX IF NOT EXISTS idx_provider_voice ON cache_entries(provider, voice_id); +-- Accelerates GET /cache/by-text exact-text lookups (the dashboard cache browser). +-- NB: substring (LIKE '%q%') still scans — a B-tree can't serve leading-wildcard +-- LIKE; add an FTS5 virtual table if substring speed ever matters. +CREATE INDEX IF NOT EXISTS idx_text ON cache_entries(text); -- Request metrics rolled up per UTC day (date-filterable via SUM, never a scan -- of cache_entries).