Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 23 additions & 7 deletions app/ai/voice/tts/dragontts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks sus

# 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 <provider> returned
# an error: <reason>"} 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
Expand All @@ -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}")
Expand Down
108 changes: 107 additions & 1 deletion app/api/routers/breeze_buddy/dragontts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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 <provider> 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"},
)
Loading
Loading