import os
import io
import re
import uuid
import time
import random
import asyncio
import tempfile
import contextlib
import logging
from collections import OrderedDict
import weakref
from core.render_trace import timed as _render_timed
import threading
import traceback
from pathlib import Path
from typing import Optional, Literal
from fastapi import APIRouter, File, Form, UploadFile, HTTPException
from fastapi.responses import JSONResponse, Response, StreamingResponse
from pydantic import BaseModel
import sqlite3
from core.db import db_conn, ensure_schema
from core.config import OUTPUTS_DIR, VOICES_DIR
import functools
from services.model_manager import (
get_model, _gpu_pool, run_on_gpu_pool_guarded, GpuJobTimeoutError,
GpuPoolBusyError,
)
from services.generation_audio import save_generation_wav
from services.audio_io import _safe_torchaudio_save
from services.binary_preflight import InvalidBinaryError
from core import event_bus
from core.render_trace import call as trace_call
from core.logging_utils import log_safe
from omnivoice.utils.voice_design import heal_design_instruct
router = APIRouter()
logger = logging.getLogger("omnivoice.generate")
_OGG_CACHE_LIMIT = 32 * 1024 * 1024
_ogg_cache: OrderedDict[tuple[str, int, int, int], bytes] = OrderedDict()
_ogg_cache_bytes = 0
_ogg_state_lock = threading.Lock()
_ogg_encode_locks = weakref.WeakValueDictionary()
@router.get('/audio/{audio_id}/quality')
def generated_audio_quality(audio_id: str):
"""Analyze a generated WAV locally, off the async event loop."""
from services.audio_quality import analyze_audio
if not re.fullmatch(r'[0-9a-f]{8}', audio_id):
raise HTTPException(status_code=404, detail='Audio file not found')
path = _safe_output_path(f'{audio_id}.wav')
if path is None or not os.path.isfile(path):
raise HTTPException(status_code=404, detail='Audio file not found')
try:
return analyze_audio(path)
except (OSError, RuntimeError, ValueError):
raise HTTPException(status_code=422, detail='Audio analysis unavailable') from None
def _ogg_cache_key(path: str) -> tuple[str, int, int, int]:
info = os.stat(path)
return path, info.st_ino, info.st_mtime_ns, info.st_size
def _cached_ogg(key: tuple[str, int, int, int]) -> bytes | None:
with _ogg_state_lock:
encoded = _ogg_cache.get(key)
if encoded is not None:
_ogg_cache.move_to_end(key)
return encoded
@router.get("/audio/{audio_id}.ogg")
@router.get("/audio/{audio_id}.opus")
async def generated_ogg_opus(audio_id: str):
"""Serve the same render as /audio/<id>.wav, encoded as Ogg/Opus."""
if not re.fullmatch(r"[0-9a-f]{8}", audio_id):
raise HTTPException(status_code=404, detail="Audio file not found")
path = _safe_output_path(f"{audio_id}.wav")
if path is None:
raise HTTPException(status_code=404, detail="Audio file not found")
try:
key = _ogg_cache_key(path)
except FileNotFoundError:
raise HTTPException(status_code=404, detail="Audio file not found") from None
encoded = _cached_ogg(key)
if encoded is not None:
return Response(encoded, media_type="audio/ogg")
with _ogg_state_lock:
encode_lock = _ogg_encode_locks.get(key)
if encode_lock is None:
encode_lock = asyncio.Lock()
_ogg_encode_locks[key] = encode_lock
async with encode_lock:
try:
if _ogg_cache_key(path) != key:
return await generated_ogg_opus(audio_id)
except FileNotFoundError:
raise HTTPException(status_code=404, detail="Audio file not found") from None
encoded = _cached_ogg(key)
if encoded is None:
from services.audio_io import encode_ogg_opus
try:
encoded = await encode_ogg_opus(path)
except asyncio.TimeoutError as exc:
logger.warning("Ogg/Opus encoding timed out")
raise HTTPException(
status_code=503, detail="Ogg/Opus encoding timed out; try again later"
) from exc
except RuntimeError as exc:
logger.warning("Ogg/Opus encoding failed: %s", exc)
raise HTTPException(status_code=503, detail=str(exc)) from exc
if len(encoded) <= _OGG_CACHE_LIMIT:
global _ogg_cache_bytes
with _ogg_state_lock:
while _ogg_cache_bytes + len(encoded) > _OGG_CACHE_LIMIT:
_ogg_cache_bytes -= len(_ogg_cache.popitem(last=False)[1])
_ogg_cache[key] = encoded
_ogg_cache_bytes += len(encoded)
return Response(encoded, media_type="audio/ogg")
_REF_UPLOAD_EXTS = frozenset({
".wav", ".mp3", ".m4a", ".flac", ".ogg", ".oga", ".opus", ".aac", ".webm",
})
def _ref_upload_suffix(filename: Optional[str]) -> str:
"""On-disk suffix for a one-shot /generate reference upload."""
ext = os.path.splitext(filename or "")[1].lower()
return ext if ext in _REF_UPLOAD_EXTS else ".wav"
class _TempReferenceLease:
"""Delete a request-owned reference once every abandoned reader drains."""
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._active = 0
self._request_done = False
self._deleted = False
def acquire(self):
with self._lock:
if self._request_done:
raise RuntimeError("reference lease acquired after request cleanup")
self._active += 1
once_lock = threading.Lock()
released = False
def release() -> None:
nonlocal released
with once_lock:
if released:
return
released = True
self._release()
return release
def _release(self) -> None:
delete = False
with self._lock:
self._active -= 1
if self._active < 0:
raise RuntimeError("reference lease released too many times")
if self._request_done and self._active == 0 and not self._deleted:
self._deleted = True
delete = True
if delete:
with contextlib.suppress(OSError):
os.remove(self.path)
def finish_request(self) -> None:
delete = False
with self._lock:
self._request_done = True
if self._active == 0 and not self._deleted:
self._deleted = True
delete = True
if delete:
with contextlib.suppress(OSError):
os.remove(self.path)
async def _run_with_reference_lease(lease, factory):
"""Hold an ad-hoc reference through one local GPU-pool dispatch."""
if lease is None:
return await factory(None)
release = lease.acquire()
abandoned = False
try:
return await factory(release)
except GpuPoolBusyError:
release()
abandoned = True
raise
except (asyncio.CancelledError, GpuJobTimeoutError):
abandoned = True
raise
finally:
if not abandoned:
release()
_DESIGN_DESCRIPTION_MAX = 2000
_DESIGN_DETAIL_MAX = 64
def _design_recipe_json(raw: Optional[str]) -> Optional[str]:
"""Validated Voice Design draft to store with a take, or None (#2389).
``design_recipe`` is display metadata from the Design page: the description
as written and the user's explicit picks. It never affects synthesis, so a
malformed or oversized value is dropped instead of failing the take. Only
known categories and bounded strings survive, and the stored JSON is
re-serialized from the parsed value — the raw form field is never
persisted. The details the description maps to are derived here with the
same mapper as ``/design/describe``, so they always match the description
this take was rendered from, even when the page's own mapping had not
landed yet.
"""
if not raw or len(raw) > 4 * _DESIGN_DESCRIPTION_MAX:
return None
import json
from core.describe_voice import CATEGORY_ORDER, parse_description
try:
value = json.loads(raw)
except ValueError:
return None
if not isinstance(value, dict):
return None
description = value.get("description")
if not isinstance(description, str) or len(description) > _DESIGN_DESCRIPTION_MAX:
return None
picks = value.get("picks")
if not isinstance(picks, dict):
return None
for category, detail in picks.items():
if (
category not in CATEGORY_ORDER
or not isinstance(detail, str)
or len(detail) > _DESIGN_DETAIL_MAX
):
return None
mapped = parse_description(description)["attrs"]
return json.dumps(
{"description": description, "picks": picks, "mapped": mapped},
ensure_ascii=False,
)
def _profile_instruct(row):
"""Validator-safe instruct for a stored profile row.
Sanitizes the persisted instruct (dropping the ``"[object Object]"``
sentinel / freeform prose that older builds saved) and, for a design row,
rebuilds the tags from ``vd_states`` when the stored value is unusable — so
a poisoned/legacy profile never 400-s generation (#550 #571 #594 #596).
"""
try:
vd = row["vd_states"]
except (KeyError, IndexError):
vd = None
return heal_design_instruct(row["instruct"], vd)
def _design_instruct_key(value):
"""Order-, case- and spacing-insensitive identity of an instruct.
Clients rebuild a design voice's tags in their own order (the sanitizer
sorts by category), so string equality would call an unchanged voice
edited.
"""
return frozenset(
item.strip().casefold()
for item in re.split(r"[,\uff0c]", str(value or ""))
if item.strip()
)
def _design_request_diverges(row, *, instruct=None, seed=None):
"""True when a request edits a design profile instead of re-rendering it.
A design profile's rendered sample *is* the voice it was saved with. A
request that sends a different instruct or seed asks for a different
voice; cloning the old sample would drown the new attributes, which is
how "Male" kept coming back female once a design was saved. An omitted
instruct or seed means "the profile's", so it never diverges; an explicit
seed on a profile saved without one asks for a specific voice the saved
sample was never rendered with, so it does.
"""
if instruct and str(instruct).strip():
requested = _design_instruct_key(instruct)
stored = {
_design_instruct_key(row["instruct"]),
_design_instruct_key(_profile_instruct(row)),
}
if requested not in stored:
return True
if seed is None:
return False
return row["seed"] is None or int(seed) != int(row["seed"])
def _conditioning_refusal(backend_cls, *, profile_id=None, has_ref_audio=False,
instruct=None, seed=None, design_recipe=None):
"""Why this request can't run on ``backend_cls``, else None.
Two conditioning mismatches would otherwise fail deep inside the engine
or, worse, render the wrong voice without a word:
* designing on an engine that declares it can't
(``supports_voice_design = False``): a request with no reference clip
that describes a voice (an instruct or a Voice Design recipe) or names
a design profile without a saved sample;
* a reference clip on an engine that declares it can't clone
(``supports_cloning = False``): an uploaded clip, or a profile whose
conditioning resolves to its saved sample. Such engines ignore
``ref_audio``, so re-rendering a saved voice there would silently use
a preset voice instead.
Plain preset-voice TTS (no reference, no description) stays allowed, and
a model-dependent ``supports_cloning`` property is judged by the loaded
model later, not here. Refusing before any model load turns both into an
actionable 422.
"""
from services.tts_backend import voice_design_support
designing = bool(
(instruct and str(instruct).strip())
or (design_recipe and str(design_recipe).strip())
)
reference = has_ref_audio and not profile_id
if profile_id:
with db_conn() as conn:
row = conn.execute(
"SELECT * FROM voice_profiles WHERE id=?", (profile_id,)
).fetchone()
if row is not None:
cond = _resolve_profile_conditioning(row, instruct=instruct, seed=seed)
reference = bool(cond["ref_audio_path"])
designing = designing or cond["kind"] == "design"
name = getattr(backend_cls, "display_name", None) or getattr(backend_cls, "id", "This engine")
if reference:
if getattr(backend_cls, "supports_cloning", True) is False:
return (
f"{name} can't use reference audio: it only speaks with its own "
"preset voices, so it can't render this voice from its sample. "
"Choose an engine that supports voice cloning (for example "
"OmniVoice)."
)
return None
if not designing or voice_design_support(backend_cls) is not False:
return None
return (
f"{name} can't design a voice from a description: it needs a reference "
"clip for the timbre. Choose an engine that supports Voice Design "
"(for example OmniVoice), or clone a voice from a reference clip instead."
)
def _resolve_profile_conditioning(row, *, ref_text=None, instruct=None,
seed=None, language=None):
"""Resolve a ``voice_profiles`` row into generation conditioning.
Extracted verbatim from /generate's inline profile-resolution block so
other synthesis routes (POST /convert) share the exact same semantics —
lock wins, ``kind`` is authoritative (0005), legacy pre-0004 rows fall
back to the is_locked/instruct inference, and #533's language fill.
Request-supplied values (``ref_text``/``instruct``/``seed``/``language``)
always win over the stored row; only gaps are filled. Returns a dict with
``ref_audio_path`` / ``ref_text`` / ``instruct`` / ``seed`` / ``language``
/ ``kind`` plus ``persist_ref_text`` — True when the caller should cache
an auto-transcribed reference transcript back onto the row (#1032).
"""
out = {
"ref_audio_path": None, "ref_text": ref_text, "instruct": instruct,
"seed": seed, "language": language, "kind": None,
"persist_ref_text": False, "language_from_profile": False,
"diverged": False,
}
try:
profile_kind = row["kind"] or "clone"
except (KeyError, IndexError):
profile_kind = "design" if (
row["instruct"] and not row["is_locked"] and not row["ref_audio_path"]
) else "clone"
out["kind"] = profile_kind
if profile_kind == "design" and _design_request_diverges(
row, instruct=instruct, seed=seed,
):
out["diverged"] = True
if not out["instruct"]:
out["instruct"] = _profile_instruct(row)
if out["seed"] is None and row["seed"] is not None:
out["seed"] = row["seed"]
elif row["is_locked"] and row["locked_audio_path"]:
out["ref_audio_path"] = os.path.join(VOICES_DIR, row["locked_audio_path"])
if not out["ref_text"]:
out["ref_text"] = row["ref_text"]
if not out["instruct"]:
out["instruct"] = _profile_instruct(row)
if out["seed"] is None and row["seed"] is not None:
out["seed"] = row["seed"]
elif profile_kind == "design":
out["ref_audio_path"] = (
os.path.join(VOICES_DIR, row["ref_audio_path"]) if row["ref_audio_path"] else None
)
if out["ref_audio_path"] and not out["ref_text"] and row["ref_text"]:
out["ref_text"] = row["ref_text"]
if not out["instruct"]:
out["instruct"] = _profile_instruct(row)
if out["seed"] is None and row["seed"] is not None:
out["seed"] = row["seed"]
elif row["instruct"] and not row["is_locked"] and not row["ref_audio_path"]:
if not out["instruct"]:
out["instruct"] = _profile_instruct(row)
if out["seed"] is None and row["seed"] is not None:
out["seed"] = row["seed"]
else:
out["ref_audio_path"] = (
os.path.join(VOICES_DIR, row["ref_audio_path"]) if row["ref_audio_path"] else None
)
if not out["ref_text"] and row["ref_text"]:
out["ref_text"] = row["ref_text"]
elif out["ref_audio_path"] and not out["ref_text"]:
out["persist_ref_text"] = True
if not out["instruct"] and row["instruct"]:
out["instruct"] = row["instruct"]
if out["seed"] is None and row["seed"] is not None:
out["seed"] = row["seed"]
explicit_auto = isinstance(language, str) and language.strip().lower() == "auto"
if explicit_auto:
out["language"] = None
if out["language"] is None and not explicit_auto:
try:
prof_lang = row["language"]
except (KeyError, IndexError):
prof_lang = None
if prof_lang and prof_lang != "Auto":
out["language"] = prof_lang
out["language_from_profile"] = True
return out
def _note_generate_progress() -> None:
"""Tell the pool guard this render just finished a unit of work (#1391).
Every multi-part render calls this after each part. A job that keeps
completing chunks is working, however slowly, and must not be abandoned as
"too heavy for the available compute" the way #1338/#1348/#1391 were —
while a job that produces nothing for a whole base budget still dies on
time. Never raises: a liveness signal that can break a render is worse
than no signal.
"""
try:
from services.model_manager import report_generate_progress
report_generate_progress()
except Exception:
pass
def _render_with_pauses(gen_span, segments, sample_rate):
"""Synthesize ``[(text, pause_ms), ...]`` spans and stitch silence between
them (issue #276).
``gen_span(text) -> torch.Tensor`` synthesizes one text span (raw model
output). A silence buffer of ``pause_ms`` is inserted after a span when
requested, matching the audio tensor's channel dims / dtype / device.
Returns the concatenated waveform. Kept model-free (``gen_span`` is injected)
so the stitching is unit-testable without loading the TTS model.
"""
import torch
items = []
for span_text, pause_ms in segments:
if span_text and span_text.strip():
items.append(("a", gen_span(span_text)))
_note_generate_progress()
if pause_ms > 0:
n = int(round(sample_rate * pause_ms / 1000.0))
if n > 0:
items.append(("s", n))
ref = next((t for kind, t in items if kind == "a"), None)
if ref is None:
total = sum(n for kind, n in items if kind == "s") or 1
return torch.zeros(total, dtype=torch.float32)
parts = []
for kind, val in items:
if kind == "a":
parts.append(val)
else:
shape = list(ref.shape)
shape[-1] = val
parts.append(torch.zeros(*shape, dtype=ref.dtype, device=ref.device))
return torch.cat(parts, dim=-1)
def _sanitize_audio(audio_out):
"""Replace non-finite samples (NaN / ±inf) with silence so a model glitch
can't produce an unreadable WAV (#629). Returns the input unchanged when it's
already finite or isn't a tensor. Never raises."""
try:
import torch
if torch.is_tensor(audio_out) and not bool(torch.isfinite(audio_out).all()):
logger.warning(
"Generated audio contained non-finite samples (NaN/inf) — "
"sanitizing to silence to keep the WAV decodable (#629)."
)
return torch.nan_to_num(audio_out, nan=0.0, posinf=0.0, neginf=0.0)
except Exception as exc:
logger.warning("Generated audio validation failed")
raise RuntimeError(
"Generated audio could not be validated. Retry the generation."
) from exc
return audio_out
@_render_timed('effects')
def _apply_effect_chain(audio_out, sample_rate, effect_preset, *, skip_mastering=False):
"""Shared post-DSP for /generate: preset validation → mastering →
effect chain → loudness normalization.
``skip_mastering`` honors a backend's ``applies_own_mastering`` flag
(issue #312): studio engines (e.g. VoxCPM2's native 48 kHz output)
opt out of the broadcast highpass + Compressor pre-stage that's tuned
for VoiceStudio's 24 kHz clone output. Loudness normalization still runs —
it's a benign peak scale. Mirrors ``_run_tts`` in openai_compat.py.
"""
from services.audio_dsp import (
EFFECT_PRESETS, apply_mastering, normalize_audio,
apply_effects_chain, get_effect_chain,
)
audio_out = _sanitize_audio(audio_out)
preset = effect_preset or "broadcast"
if preset not in EFFECT_PRESETS:
raise ValueError(
f"Unknown effect preset: {preset!r}. "
f"Valid: {list(EFFECT_PRESETS.keys())}"
)
if preset == "raw":
return audio_out
if not skip_mastering:
audio_out = apply_mastering(audio_out, sample_rate=sample_rate)
chain = get_effect_chain(preset)
if chain:
audio_out = apply_effects_chain(
audio_out, sample_rate=sample_rate, chain=chain,
)
return normalize_audio(audio_out, target_dBFS=-2.0)
def _safe_exc_text(e: BaseException) -> str:
"""``f"{type(e).__name__}: {e}"`` — the house style used for
unrecognized-error formatting throughout the backend (grep
``type(e).__name__`` in settings.py / asr_backend.py / model_manager.py
/ engines.py) — with a guard against leaking a raw container repr.
#977: an AssertionError raised deep inside a vendored dependency
(mlx-audio's Kokoro pipeline) had ``.args`` shaped like
``('du', {'a': 'American English', ...})`` — a tuple containing a dict.
``str(e)`` on that renders the WHOLE table straight into the user-facing
message. Any engine's ``generate()`` can raise something shaped like
this (not just Kokoro), so guard generically: if any element of
``e.args`` is a container rather than a plain string, don't interpolate
``str(e)`` at all — name the exception type and point at the log
instead.
"""
args = getattr(e, "args", ())
if any(isinstance(a, (dict, list, tuple, set, frozenset)) for a in args):
return f"{type(e).__name__} — see Settings → Logs → Backend for details"
return f"{type(e).__name__}: {e}"
def _exception_chain(e):
"""Yield ``e`` plus every ``__cause__``/``__context__`` beneath it
(cycle-safe). Engines and hub libraries routinely wrap the original
transport/allocator error, so classification must look at the whole
chain, not just the outermost message."""
seen = set()
stack = [e]
while stack:
exc = stack.pop()
if exc is None or id(exc) in seen:
continue
seen.add(id(exc))
yield exc
stack.append(exc.__cause__)
stack.append(exc.__context__)
_NETWORK_EXC_NAMES = frozenset({
"ConnectError", "ConnectTimeout", "ReadTimeout", "ReadError",
"WriteError", "WriteTimeout", "PoolTimeout", "NetworkError",
"TransportError", "RemoteProtocolError", "ProxyError", "CloseError",
"ConnectionError", "ChunkedEncodingError", "MaxRetryError",
"NewConnectionError", "ProtocolError",
"ConnectionResetError", "ConnectionAbortedError", "ConnectionRefusedError",
"LocalEntryNotFoundError",
})
_NETWORK_MSG_SIGNATURES = (
"client has been closed",
"cannot send a request",
"connection error",
"connection reset",
"read timed out",
"max retries exceeded",
"temporary failure in name resolution",
"name or service not known",
"getaddrinfo failed",
)
_TLS_DROP_SIGNATURES = (
"unexpected_eof_while_reading",
"eof occurred in violation of protocol",
)
def _is_network_failure(e) -> bool:
"""True iff the failure (anywhere in its chain) is an HTTP-client
lifecycle / network-transport error — e.g. a first-use model download
from the HF Hub dying mid-generation (#880)."""
for exc in _exception_chain(e):
if type(exc).__name__ in _NETWORK_EXC_NAMES:
return True
low = str(exc).lower()
if any(sig in low for sig in _NETWORK_MSG_SIGNATURES):
return True
if "ssl" in low and any(sig in low for sig in _TLS_DROP_SIGNATURES):
return True
return False
_OOM_MSG_SIGNATURES = (
"out of memory",
"not enough memory",
"cannot allocate memory",
"std::bad_alloc",
"cublas_status_alloc_failed",
"cuda_error_out_of_memory",
"paging file is too small",
)
def _is_oom_failure(e) -> bool:
"""True iff the failure (anywhere in its chain) actually looks like an
out-of-memory condition — the only case where the Flush hint is honest."""
for exc in _exception_chain(e):
if isinstance(exc, MemoryError):
return True
if type(exc).__name__ == "OutOfMemoryError":
return True
low = str(exc).lower()
if any(sig in low for sig in _OOM_MSG_SIGNATURES):
return True
return False
_CONFIG_MSG_SIGNATURES = (
"not set. point it to",
"no model.onnx found in",
"not configured",
"venv not found. set",
"unavailable: omnivoice_",
)
_CONFIG_ENV_RE = re.compile(r"omnivoice_[a-z0-9_]+")
def _is_config_failure(e) -> bool:
"""True iff the failure is a *configuration* problem — a required engine
model path / env var that isn't set (or points nowhere) — rather than a
runtime fault. The remedy is to set the value, never to Flush VRAM."""
for exc in _exception_chain(e):
low = str(exc).lower()
if any(sig in low for sig in _CONFIG_MSG_SIGNATURES):
return True
if _CONFIG_ENV_RE.search(low) and (
"not set" in low or "point it to" in low or "set omnivoice_" in low
):
return True
return False
_TIMEOUT_EXC_NAMES = frozenset({
"TimeoutError", "GpuJobTimeoutError", "FuturesTimeoutError",
})
_TIMEOUT_MSG_SIGNATURES = (
"timed out",
"timeout expired",
"exceeded its time budget",
)
def _is_timeout_failure(e) -> bool:
"""True iff the generation ran out of *time* rather than failing (#1368).
A bare ``TimeoutError`` used to fall through to the unrecognized-error
catch-all, so the user was told to "retry once and report it with the full
trace" for the one failure mode whose cause is fully known and whose
message is usually EMPTY — ``TimeoutError:`` with nothing after the colon
tells them nothing at all.
Deliberately checked before the OOM branch: a job killed at its deadline is
not an allocation failure, and Flush is the wrong remedy for it.
"""
for exc in _exception_chain(e):
if isinstance(exc, TimeoutError) or type(exc).__name__ in _TIMEOUT_EXC_NAMES:
return True
low = str(exc).lower()
if any(sig in low for sig in _TIMEOUT_MSG_SIGNATURES):
if "read timed out" in low:
continue
return True
return False
def _is_media_process_launch_failure(exc: BaseException) -> bool:
"""Identify an ffmpeg/ffprobe launch ENOENT without guessing from a file name."""
if not isinstance(exc, FileNotFoundError):
return False
traceback_cursor = exc.__traceback__
if traceback_cursor is None:
return False
while traceback_cursor.tb_next is not None:
traceback_cursor = traceback_cursor.tb_next
origin_module = traceback_cursor.tb_frame.f_globals.get("__name__", "")
if origin_module != "subprocess" and not origin_module.startswith("asyncio."):
return False
filename = getattr(exc, "filename", None)
if not filename:
return "[winerror 2]" in str(exc).lower()
return os.path.basename(str(filename)).lower() in {
"ffmpeg", "ffmpeg.exe", "ffprobe", "ffprobe.exe",
}
def _oom_friendly_reraise(e):
"""Best-effort cache flush + the user-facing OOM hint shared by both
inference paths."""
import gc
from services.model_manager import release_device_cache
gc.collect()
release_device_cache()
from services.model_manager import _is_compile_runtime_failure
if _is_compile_runtime_failure(e):
raise RuntimeError(
f"TTS engine hit a torch.compile/Triton error (not out of memory). "
f"Disable torch.compile in Settings → Performance, use the Flush "
f"button to reload the model, then regenerate. Underlying error: {e}"
) from e
es = str(e)
for _exc in _exception_chain(e):
if _is_media_process_launch_failure(_exc):
raise RuntimeError(
"A required media program couldn't be launched. Open "
"Settings → Audio tools and use "
"Download/Repair for the media engine, then retry. If Audio "
"tools is already ready, repair the selected TTS engine and "
f"restart VoiceStudio. Underlying error: {_safe_exc_text(_exc)}"
) from e
if isinstance(e, PermissionError) or "Permission denied" in es or "Errno 13" in es:
raise RuntimeError(
f"A required engine binary couldn't be executed (permission denied). "
f"This usually means a bundled binary lost its execute bit — reinstall, "
f"or run `chmod +x` on the engine binary named in the error. "
f"Underlying error: {e}"
) from e
if "ffmpeg returned error" in es or "Decoding failed" in es or "Invalid data found" in es:
raise RuntimeError(
f"The engine produced unreadable audio (a decode step failed) — this is "
f"usually a transient glitch. Use the Flush button to reload the model, "
f"then regenerate. Underlying error: {e}"
) from e
_low = es.lower()
if ("unsupported instruct items" in _low
or "conflicting instruct items" in _low
or "in a single instruct" in _low):
raise ValueError(es) from e
if "[winerror 193]" in _low or "is not a valid win32 application" in _low:
raise RuntimeError(
f"A native component (a DLL / .pyd / .exe — e.g. torch, ffmpeg, or an "
f"engine binary) is corrupt or built for the wrong architecture "
f"([WinError 193]). Reinstall or repair that component — the Flush "
f"button won't help here. Underlying error: {e}"
) from e
if ("[winerror 4551]" in _low or "[winerror 1260]" in _low
or "application control policy" in _low):
raise RuntimeError(
f"Windows blocked a file VoiceStudio needs from running — an "
f"Application Control policy (Smart App Control, WDAC, or "
f"AppLocker) refused to load it. On a personal PC: Windows "
f"Security → App & browser control → Smart App Control → Off "
f"(note Windows only lets you turn it off once — re-enabling "
f"needs a Windows reset), then restart VoiceStudio. On a managed/"
f"work PC ask IT to allow the VoiceStudio install folder. The Flush "
f"button won't help. Underlying error: {e}"
) from e
if "libsndfile" in _low or "writing the audio file failed" in _low:
raise RuntimeError(
f"An audio file couldn't be read or written (libsndfile failed at "
f"the OS level). This is a file/disk problem, not a memory one: "
f"check the drive isn't full, the output and temp folders exist "
f"and are writable, and that antivirus or OneDrive isn't locking "
f"them (add a VoiceStudio exclusion if you use one). If it happens "
f"only with one reference clip, re-import that clip. Underlying "
f"error: {e}"
) from e
if "no kernel image is available" in _low:
raise RuntimeError(
f"Your GPU isn't supported by the installed PyTorch build (CUDA can't "
f"launch kernels for its compute capability). Switch the compute device "
f"to CPU in Settings, or install a matching PyTorch (e.g. a cu128 build "
f"for newer GPUs). The Flush button won't help. Underlying error: {e}"
) from e
if isinstance(e, BrokenPipeError) or "broken pipe" in _low or "errno 32" in _low:
raise RuntimeError(
f"The backend lost its output pipe mid-generation — the desktop app "
f"that launched it closed or relaunched ([Errno 32] Broken pipe). "
f"Restart the app and try again; the Flush button won't help here. "
f"Underlying error: {e}"
) from e
if _is_network_failure(e):
raise RuntimeError(
f"A model download or network call failed mid-generation (usually "
f"the engine fetching its model files on first use). This is a "
f"network problem, not a memory problem — flushing VRAM won't "
f"help. Retry the generation; if it keeps failing, check your "
f"internet connection and any HF_ENDPOINT/mirror setting. "
f"Underlying error: {e}"
) from e
if _is_config_failure(e):
raise RuntimeError(
f"This TTS engine isn't set up yet — it needs a model path or "
f"environment variable that isn't configured, so nothing was "
f"generated. Set it as the underlying error describes (it names the "
f"exact variable and what to point it at), then restart VoiceStudio — "
f"or pick a ready engine in Model Catalogue. This is a setup "
f"problem, not a memory one. Underlying error: {e}"
) from e
if _is_timeout_failure(e):
_tail = f" Underlying error: {_safe_exc_text(e)}" if str(e).strip() else ""
raise RuntimeError(
"The engine hit its time limit before finishing, so generation was "
"stopped. Nothing is broken and flushing memory won't help. The "
"usual causes are a first-use model download still in progress "
"(retry once it finishes — it resumes), a very long input, or an "
"engine running on CPU. Shorter text, or raising "
"OMNIVOICE_GENERATE_TIMEOUT_S, will get it through."
+ _tail
) from e
_low_1455 = str(e).lower()
if "paging file is too small" in _low_1455 or (
"1455" in _low_1455 and ("winerror" in _low_1455 or "os error" in _low_1455)
):
from core.failure import _HINTS
raise RuntimeError(
"Windows ran out of virtual memory while mapping the model — its "
"paging file is smaller than the model needs. This is not your RAM "
"being full, it is not a network problem, and Flush cannot help. "
+ _HINTS["WINDOWS_PAGING_FILE_TOO_SMALL"]
+ f" Underlying error: {e}"
) from e
from core.failure import _HINTS, classify
if classify(str(e)) == "POCKETTTS_GATED_WEIGHTS":
raise RuntimeError(
"PocketTTS voice cloning uses gated weights this install can't "
"download yet; its built-in voices still work. "
+ _HINTS["POCKETTTS_GATED_WEIGHTS"]
+ f" Underlying error: {_safe_exc_text(e)}"
) from e
if classify(str(e)) == "HF_AUTH_FAILED":
raise RuntimeError(
"This model needs files from a gated Hugging Face repository that "
"this install can't access yet. " + _HINTS["HF_AUTH_FAILED"]
+ f" Underlying error: {_safe_exc_text(e)}"
) from e
if _is_oom_failure(e):
raise RuntimeError(
f"TTS engine stopped mid-generation. This usually means it ran out of memory. "
f"Try the Flush button to reload the model, then regenerate. Underlying error: {e}"
) from e
raise RuntimeError(
f"TTS engine stopped mid-generation with an error VoiceStudio doesn't "
f"recognize. Retry once; if it keeps failing, please report it with "
f"the full trace. Underlying error: {_safe_exc_text(e)}"
) from e
def _generate_timeout_s(
text: str,
*,
engine: object = None,
execution_device=None,
min_vram_gb=0.0,
hardware_family=None,
vram_gb=None,
) -> float:
"""Wall-clock budget for one generate, scaled to the request.
Thin alias for the canonical helper, which moved to
``services.model_manager.generate_timeout_s`` (#1190) so /v1/audio/speech,
batch, dub and archetype previews share it instead of each re-deriving (or,
as they did, silently keeping the flat 300s).
``min_vram_gb`` is the engine's declared VRAM floor. A GPU below it pages to
system RAM and renders slower than this machine's CPU, so it must not be
budgeted as fast hardware (#1804) — the same figure the dispatch already
hands the guard so a timeout message can name the card (#1226/#1222).
"""
from services.model_manager import generate_timeout_s
return generate_timeout_s(
text,
engine=engine,
execution_device=execution_device,
min_vram_gb=min_vram_gb,
hardware_family=hardware_family,
vram_gb=vram_gb,
)
def _run_inference(
model, text, language, ref_audio_path, ref_text, instruct, duration,
num_step, guidance_scale, speed, t_shift, denoise,
postprocess_output, layer_penalty_factor, position_temperature,
class_temperature, used_seed, effect_preset="broadcast",
max_chunk_chars=None, crossfade_ms=None, *, dropped_sink=None,
):
import torch
try:
if used_seed is not None:
torch.manual_seed(used_seed)
kwargs = {}
if t_shift is not None: kwargs["t_shift"] = t_shift
if layer_penalty_factor is not None: kwargs["layer_penalty_factor"] = layer_penalty_factor
if position_temperature is not None: kwargs["position_temperature"] = position_temperature
if class_temperature is not None: kwargs["class_temperature"] = class_temperature
sr = model.sampling_rate if hasattr(model, 'sampling_rate') else 24000
from services.tts_backend import generate_with_cached_ref
def _gen(gen_text, gen_duration):
"""One generate call for this request's voice, reference encoded once."""
return trace_call("synthesis", generate_with_cached_ref,
model, ref_audio=ref_audio_path, ref_text=ref_text,
text=gen_text, language=language, instruct=instruct,
duration=gen_duration, num_step=num_step,
guidance_scale=guidance_scale, speed=speed, denoise=denoise,
postprocess_output=postprocess_output, **kwargs
)
from omnivoice.utils.text import parse_pause_markers
segments = parse_pause_markers(text)
has_pause = len(segments) > 1 or (segments and segments[0][1] > 0)
if has_pause:
def _gen_span(span_text):
return _gen(span_text, None)[0]
audio_out = _render_with_pauses(_gen_span, segments, sr)
else:
from services.chunked_tts import (
DEFAULT_CROSSFADE_MS, DEFAULT_MAX_CHUNK_CHARS,
concatenate_audio_chunks, split_text_into_chunks,
)
_max_chars = DEFAULT_MAX_CHUNK_CHARS if max_chunk_chars is None else max_chunk_chars
_xfade_ms = DEFAULT_CROSSFADE_MS if crossfade_ms is None else crossfade_ms
text_chunks = split_text_into_chunks(text, _max_chars)
if len(text_chunks) > 1:
parts = []
for i, chunk_text in enumerate(text_chunks):
if used_seed is not None:
torch.manual_seed(used_seed + i)
parts.append(_gen(chunk_text, None)[0])
_note_generate_progress()
audio_out = concatenate_audio_chunks(parts, sr, _xfade_ms,
texts=text_chunks,
sink=dropped_sink)
else:
audio_out = _gen(text, duration)[0]
return _apply_effect_chain(audio_out, sr, effect_preset)
except ValueError as e:
raise e
except Exception as e:
_oom_friendly_reraise(e)
def _run_backend_inference(
backend, text, language, ref_audio_path, ref_text, instruct, duration,
num_step, guidance_scale, speed, denoise, postprocess_output,
used_seed, effect_preset="broadcast",
max_chunk_chars=None, crossfade_ms=None, *, t_shift=None,
layer_penalty_factor=None, position_temperature=None,
class_temperature=None, dropped_sink=None,
):
"""Engine-aware twin of :func:`_run_inference` (issue #312).
Runs the request through a pluggable ``TTSBackend`` adapter instead of the
VoiceStudio model directly. A crash-isolated OmniVoice proxy advertises
``supports_native_omnivoice_controls`` and receives the same advanced
controls and per-call seed as the native path; other adapters keep the
narrower protocol unchanged.
"""
import torch
try:
if used_seed is not None:
torch.manual_seed(used_seed)
if language and language.lower() == "auto":
language = None
gen_kwargs = dict(
language=language, ref_audio=ref_audio_path, ref_text=ref_text,
instruct=instruct, num_step=num_step, guidance_scale=guidance_scale,
speed=speed, denoise=denoise, postprocess_output=postprocess_output,
)
native_proxy = bool(
getattr(backend, "supports_native_omnivoice_controls", False)
)
forwards_seed = native_proxy or getattr(backend, "supports_generation_seed", False)
if native_proxy:
gen_kwargs.update({
key: value for key, value in {
"t_shift": t_shift,
"layer_penalty_factor": layer_penalty_factor,
"position_temperature": position_temperature,
"class_temperature": class_temperature,
}.items() if value is not None
})
sr = backend.sample_rate
from omnivoice.utils.text import parse_pause_markers
segments = parse_pause_markers(text)
has_pause = len(segments) > 1 or (segments and segments[0][1] > 0)
if has_pause:
first_span = True
def _gen_span(span_text):
nonlocal first_span
span_kwargs = dict(gen_kwargs)
if forwards_seed and first_span and used_seed is not None:
span_kwargs["seed"] = used_seed
first_span = False
return trace_call("synthesis", backend.generate, span_text, duration=None, **span_kwargs)
audio_out = _render_with_pauses(_gen_span, segments, sr)
else:
from services.chunked_tts import (
DEFAULT_CROSSFADE_MS, DEFAULT_MAX_CHUNK_CHARS,
concatenate_audio_chunks, split_text_into_chunks,
)
_max_chars = DEFAULT_MAX_CHUNK_CHARS if max_chunk_chars is None else max_chunk_chars
_xfade_ms = DEFAULT_CROSSFADE_MS if crossfade_ms is None else crossfade_ms
text_chunks = split_text_into_chunks(text, _max_chars)
if len(text_chunks) > 1:
parts = []
for i, chunk_text in enumerate(text_chunks):
if used_seed is not None:
torch.manual_seed(used_seed + i)
chunk_kwargs = dict(gen_kwargs)
if forwards_seed and used_seed is not None:
chunk_kwargs["seed"] = used_seed + i
parts.append(trace_call("synthesis", backend.generate,
chunk_text, duration=None, **chunk_kwargs
))
_note_generate_progress()
audio_out = concatenate_audio_chunks(parts, sr, _xfade_ms,
texts=text_chunks,
sink=dropped_sink)
else:
if forwards_seed and used_seed is not None:
gen_kwargs["seed"] = used_seed
audio_out = trace_call("synthesis", backend.generate, text, duration=duration, **gen_kwargs)
return _apply_effect_chain(
audio_out, sr, effect_preset,
skip_mastering=getattr(backend, "applies_own_mastering", False),
)
except ValueError as e:
raise _language_rejection_or(e, backend, language)
except Exception as e:
rewritten = _language_rejection_or(e, backend, language)
if rewritten is not e:
raise rewritten from e
_oom_friendly_reraise(e)
_LANGUAGE_REJECTION_SIGNATURES = (
"invalid language code",
"language not supported",
"language is not supported",
"unsupported language code",
)
_SELF_DESCRIBING_LANGUAGE_REJECTIONS = (
"doesn't support language",
"does not support language",
)
_LANGUAGE_REJECTION_RE = re.compile(
r"unsupported language\s*[:=]|unsupported language\s*['\"]|"
r"unsupported language\s*$",
re.IGNORECASE | re.MULTILINE,
)
def _is_language_rejection(text: str) -> bool:
"""True when an engine failure is about the LANGUAGE it was handed.
Matched on the message, not the type: the engines multiplex third-party
libraries that each raise their own class. Covers the self-describing
wordings too — #1257's rewrite skips those, but #2156 still needs to know a
language was refused so it can say where that language came from.
"""
low = text.lower()
return (
any(sig in low for sig in _LANGUAGE_REJECTION_SIGNATURES)
or any(sig in low for sig in _SELF_DESCRIBING_LANGUAGE_REJECTIONS)
or bool(_LANGUAGE_REJECTION_RE.search(text))
)
def _profile_language_rejection_detail(exc: BaseException, language) -> str:
"""The 400 body for a language the *voice profile* supplied, not the user.
#2156: the UI omits `language` while its picker reads "Auto", and #533
fills that gap from the selected profile. When the active engine can't
speak the profile's language the engine's own message tells the user to
"leave language as 'Auto'" — which is exactly what they did, so the advice
cannot be acted on. Name the real source and the remedies that exist.
"""
return (
f"This voice profile is saved with the language '{language}', and the "
f"active engine can't speak it. The language picker being on \"Auto\" "
f"does not override that — Auto fills the language in from the "
f"profile. Set this voice profile's language to one the engine "
f"supports, pick a supported language explicitly for this render, or "
f"switch engine in Model Catalogue (the VoiceStudio engine has the "
f"widest coverage). Engine's own message: {exc}"
)
def _language_rejection_payload(exc, language, *, from_profile):
"""Stable, non-retryable error metadata for both response transports."""
from core.public_errors import stream_failure
failure = stream_failure("invalid_request")
failure["terminal"] = True
if from_profile:
failure.update(
code="profile_language_rejected", language=language,
detail=_profile_language_rejection_detail(_root_language_error(exc), language),
)
return failure
def _language_rejection_http_error(exc: BaseException, language, *, from_profile):
"""The 400 a refused language deserves, wherever the refusal was raised.
A language an engine cannot speak is never retryable — not by waiting, and
not by re-running the same request on another machine. Built here so the
local (`ValueError`) and remote (`RemoteJobFailed`) handlers cannot drift:
#2156 shipped the profile-aware branch on the local path only, and a remote
render kept answering with a retryable 503 that offered "run it on this
machine instead", which cannot help.
"""
root = _root_language_error(exc)
detail = (
_profile_language_rejection_detail(root, language)
if from_profile else str(exc)
)
if from_profile:
detail = {"code": "profile_language_rejected", "language": language, "message": detail}
return HTTPException(status_code=400, detail=detail)
def _language_rejection_or(e: BaseException, backend, language):
"""``e`` rewritten with engine context when it's a language rejection.
Returns ``e`` unchanged otherwise, so this is safe to wrap any failure in.
Deliberately narrower than :func:`_is_language_rejection`: a message that
already names its engine and the languages it supports is left alone rather
than nested inside a second "Engine's own message:".
"""
text = str(e)
low = text.lower()
if any(sig in low for sig in _SELF_DESCRIBING_LANGUAGE_REJECTIONS):
return e
if not any(sig in low for sig in _LANGUAGE_REJECTION_SIGNATURES) and not (
_LANGUAGE_REJECTION_RE.search(text)
):
return e
engine = getattr(backend, "display_name", None) or getattr(
type(backend), "id", type(backend).__name__
)
requested = f" '{language}'" if language else ""
rewritten = ValueError(
f"The {engine} engine can't speak{requested}. VoiceStudio offers every "
f"language its default engine supports, but each engine covers a "
f"different set — pick one this engine supports, or switch engine in "
f"Model Catalogue (the VoiceStudio engine has the widest coverage) "
f"and generate again. Engine's own message: {e}"
)
rewritten.engine_language_error = e
return rewritten
def _root_language_error(exc: BaseException) -> BaseException:
"""The engine's own rejection, unwrapping :func:`_language_rejection_or`."""
return getattr(exc, "engine_language_error", exc)
def _persist_profile_ref_text(profile_id: str, ref_text: str) -> None:
"""Cache an auto-transcribed reference transcript onto its profile row.
#1032 perf regression: profiles saved without a transcript re-ran a FULL
ASR model load + transcribe on every /generate (the #308 auto-transcribe
path). Persisting the first transcript makes subsequent generates read it
from the row like a user-entered one. The guarded UPDATE only ever fills
an empty column — it can never overwrite a transcript the user typed or a
lock wrote — and a failure is logged, never raised (best-effort, same
contract as the transcribe itself)."""
try:
with db_conn() as conn:
updated = conn.execute(
"UPDATE voice_profiles SET ref_text=? "
"WHERE id=? AND (ref_text IS NULL OR ref_text='')",
(ref_text, profile_id),
).rowcount
if updated:
event_bus.emit("profiles", {"action": "updated", "id": profile_id})
except Exception as e:
logger.warning(
"could not persist auto-transcribed ref_text onto profile %s: %s",
log_safe(profile_id), log_safe(e),
)
async def _finalize_generation(
audio_tensor, sample_rate, *, text, history_mode, ref_audio_path,
language, instruct, resolved_profile_id, used_seed, start_time,
already_marked=False, design_recipe=None, wav_bits=16, include_wav_bytes=False,
):
"""Shared tail of a successful generation: watermark → save WAV →
history row (self-healing) → retention prune → event emit.
Used verbatim by both the classic whole-file response path and the
streaming-preview path (``stream=true``), so the on-disk artifact —
watermark, filename, history row, retention behavior — is identical
regardless of how the audio was delivered to the client.
``already_marked`` is for audio that arrives provenance-marked: a remote
worker marks at the tensor stage before it encodes (with ``force=True``,
so the *requesting* user's preference governs, not the GPU owner's), and
embedding a second AudioSeal payload over the first degrades detection of
both. The take users keep carries exactly one whole-take mark either way.
Returns ``(watermarked_tensor, meta)`` where ``meta`` carries
``id`` / ``filename`` / ``duration`` / ``gen_time``.
"""
if not already_marked:
from services.watermark import mark_synthetic_async
audio_tensor = await mark_synthetic_async(
audio_tensor, sample_rate, context="generate.finalize",
)
gen_time = round(time.time() - start_time, 2)
for _ in range(16):
audio_id = str(uuid.uuid4())[:8]
audio_filename = f"{audio_id}.wav"
audio_path = os.path.join(OUTPUTS_DIR, audio_filename)
try:
reservation = os.open(audio_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
continue
os.close(reservation)
break
else:
raise OSError("Could not reserve a unique recording filename")
def _save_take():
try:
save_generation_wav(audio_path, audio_tensor, sample_rate, bits=wav_bits)
return Path(audio_path).read_bytes() if include_wav_bytes else None
except Exception:
Path(audio_path).unlink(missing_ok=True)
raise
saving = asyncio.create_task(asyncio.to_thread(_save_take))
try:
response_bytes = await asyncio.shield(saving)
except asyncio.CancelledError:
while not saving.done():
try:
await asyncio.shield(saving)
except asyncio.CancelledError:
continue
except Exception:
break
try:
saving.result()
except Exception:
pass
Path(audio_path).unlink(missing_ok=True)
raise
audio_dur = round(audio_tensor.shape[-1] / sample_rate, 2)
take_mode = history_mode or ("clone" if ref_audio_path else "design")
if take_mode != "design":
design_recipe = None
def _write_history():
with db_conn() as conn:
conn.execute(
"INSERT INTO generation_history (id, text, mode, language, instruct, profile_id, audio_path, duration_seconds, generation_time, seed, design_recipe, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(audio_id, text[:200], take_mode,
language or "Auto", instruct or "", resolved_profile_id,
audio_filename, audio_dur, gen_time, used_seed, design_recipe, time.time())
)
try:
_write_history()
except sqlite3.OperationalError as e:
logger.warning("generation history write failed (%s); healing schema + retrying", e)
try:
ensure_schema()
_write_history()
except Exception as e2:
logger.warning("history write still failed after schema heal; returning audio anyway: %s", e2)
except Exception as e:
logger.warning("generation history write failed; returning audio anyway: %s", e)
try:
_prune_history_over_cap(keep_id=audio_id)
except Exception as e:
logger.warning("history retention prune failed (non-fatal): %s", e)
event_bus.emit("generation_history", {"action": "created", "id": audio_id})
try:
from core.analytics import capture as _ph
_ph("speech_generated", {
"mode": history_mode,
"language": language or "auto",
"duration_seconds": audio_dur,
"gen_time_seconds": gen_time,
"text_length": len(text or ""),
"has_profile": bool(resolved_profile_id),
})
except Exception:
logger.warning("analytics: speech_generated capture failed", exc_info=True)
return audio_tensor, {
"id": audio_id,
"filename": audio_filename,
"duration": audio_dur,
"gen_time": gen_time,
**({"_wav_bytes": response_bytes} if include_wav_bytes else {}),
}
def _pcm16_b64(wav_tensor) -> str:
"""Mono 16-bit little-endian PCM, base64-encoded — the streaming-preview
wire format (same conversion as /ws/tts). N-D tensors take channel 0."""
import base64
import torch
pcm = (wav_tensor * 32767).clamp(-32768, 32767).to(torch.int16)
while pcm.ndim > 1:
pcm = pcm[0]
return base64.b64encode(pcm.cpu().numpy().tobytes()).decode("ascii")
_REMOTE_OP = "tts"
_REMOTE_PHASE_LABELS = {
"queued": "queued on {target}",
"loading": "loading model on {target}",
"running": "generating on {target}",
"uploading": "receiving audio from {target}",
}
class _LocalDecision:
"""Stand-in for ``worker.routing.Decision`` meaning "run here".
Used only when the gateway cannot be imported at all, so a build without
it still renders instead of 500-ing.
"""
remote = False
worker_id = None
label = "Local"
reason = ""
_LOCAL_DECISION = _LocalDecision()
def _routing_decision():
"""Local or remote for this request — resolved once, never re-asked.
Asked once because the target is user-settable at any moment: a decision
that flipped between prewarm and dispatch would either warm an engine
nothing will use or dispatch remotely after paying a local cold load.
"""
try:
from services import gpu_gateway
return gpu_gateway.decide(_REMOTE_OP)
except Exception:
logger.debug("remote routing unavailable; running locally", exc_info=True)
return _LOCAL_DECISION
def _remote_only_local_call(target_label, reason=""):
"""The local branch of a render whose local half was deliberately skipped.
``gpu_gateway.run`` always takes a local callable — it is where rule 1
(pre-dispatch unavailability) lands. But this route skips every local
preparation step once the decision is remote, precisely so a job bound for
the 4090 does not first load gigabytes here, so there is no local render
left to fall back to.
The causes rule 1 actually covers — worker offline, disabled, not
approved, breaker open, remote workers switched off — are already answered
by ``decide()`` BEFORE that skip, and come back as a local decision with a
named reason. What is left is the narrow window where dispatch itself is
refused (a full queue, a task dropped between submit and wait). Saying so
and offering the local re-run is honest; silently returning nothing is not.
"""
from services.gpu_gateway import RemoteJobFailed
def _refuse():
raise RemoteJobFailed(
reason or f"{target_label} could not take this render",
worker_label=target_label,
code="REMOTE_NOT_DISPATCHED",
hint="Run it on this machine instead, or pick another GPU.",
)
return _refuse
def _remote_progress_frame(state, target):
"""One gateway ``on_state`` payload → the NDJSON event the UI renders."""
phase = str((state or {}).get("phase") or "running")
try:
pct = max(0, min(100, round(float((state or {}).get("progress") or 0.0) * 100)))
except (TypeError, ValueError):
pct = 0
detail = _REMOTE_PHASE_LABELS.get(phase, _REMOTE_PHASE_LABELS["running"])
detail = detail.format(target=target)
if phase == "running" and pct:
detail = f"{detail} ({pct}%)"
return {
"type": "progress", "stage": phase, "percent": pct,
"target": target, "detail": detail,
}
def _apply_routing_headers(headers, engine_notice, decision):
"""Say where this render ran, on the notice channel that already exists.
``X-OmniVoice-Routing`` / ``-Routing-Reason`` are already set for the #21
engine routing gate and already consumed as a de-duped one-time toast, so
"this ran on gpu2" and "your 4090 was asleep, this ran here" travel the
same wire rather than inventing a second one.
The engine notice wins on a local render: "the engine fell back to CPU"
explains the slowness the user is looking at, while the worker notice for
a local render is the quieter of the two. A remote render has no engine
notice at all — that gate answers for THIS host, and this host did nothing.
"""
from services.engine_routing import header_safe_reason
notice = engine_notice
if decision is not None:
try:
from services.gpu_gateway import notice_for
worker_notice = notice_for(decision)
except Exception:
worker_notice = None
if worker_notice and (getattr(decision, "remote", False) or not notice):
notice = worker_notice
if not notice:
return headers
headers["X-OmniVoice-Routing"] = notice[0]
safe = header_safe_reason(notice[1]) if notice[1] else ""
if safe:
headers["X-OmniVoice-Routing-Reason"] = safe
return headers
@router.get("/generate/budget")
def generate_budget(engine: Optional[str] = None):
"""Active generate budgets, so the UI's backstop follows operator overrides.
``engine`` (default: the active engine) selects the local route the
CPU-ceiling hint is reported for."""
from services.model_manager import generate_budget_s
return generate_budget_s(engine)
@router.post("/generate")
async def generate_speech(
text: str = Form(...),
language: Optional[str] = Form(None),
ref_audio: Optional[UploadFile] = File(None),
ref_text: Optional[str] = Form(None),
instruct: Optional[str] = Form(None),
design_recipe: Optional[str] = Form(None),
duration: Optional[float] = Form(None),
num_step: Optional[int] = Form(None),
guidance_scale: float = Form(2.0),
speed: float = Form(1.0),
t_shift: Optional[float] = Form(None),
denoise: bool = Form(True),
postprocess_output: Optional[bool] = Form(None),
layer_penalty_factor: Optional[float] = Form(None),
position_temperature: Optional[float] = Form(None),
class_temperature: Optional[float] = Form(None),
profile_id: Optional[str] = Form(None),
seed: Optional[int] = Form(None),
effect_preset: str = Form("broadcast"),
wav_bits: Literal["16", "24", "32"] = Form("16"),
engine: Optional[str] = Form(None),
max_chunk_chars: int = Form(800, ge=0),
crossfade_ms: int = Form(50, ge=0, le=1000),
pronounce: bool = Form(True),
stream: bool = Form(False),
):
import unicodedata
text = unicodedata.normalize("NFC", text)
from services.tts_backend import (
OmniVoiceBackend, _mask_hf_tokens, active_backend_id, get_backend_class,
)
engine_id = engine or active_backend_id()
from services.performance_profiles import tts_defaults
sampling_defaults = tts_defaults(engine_id)
if num_step is None:
num_step = sampling_defaults.get("num_step", 16)
if postprocess_output is None:
postprocess_output = sampling_defaults.get("postprocess_output", True)
try:
backend_cls = get_backend_class(engine_id)
except ValueError:
raise HTTPException(
status_code=400,
detail=(
f"Unknown TTS engine: {engine_id!r}. "
"See GET /engines/tts for the list of valid engine ids."
),
)
_refused = _conditioning_refusal(
backend_cls, profile_id=profile_id, has_ref_audio=ref_audio is not None,
instruct=instruct, seed=seed, design_recipe=design_recipe,
)
if _refused:
raise HTTPException(status_code=422, detail=_refused)
from core.run_sentinel import touch_activity
touch_activity("generate", engine_id)
_decision = _routing_decision()
_remote = bool(getattr(_decision, "remote", False))
_target_label = getattr(_decision, "label", "") or "the chosen worker"
_model = None
_backend = None
_engine_min_vram_gb = getattr(backend_cls, "min_vram_gb", 0.0)
_routing_notice = None
_routing = {"effective_device": None}
_routing_hardware_family = None
_routing_vram_gb = None
if not _remote:
from services.engine_memory import evict_other_tts_engines
await evict_other_tts_engines(engine_id)
try:
from services.memory_budget import log_if_low
log_if_low(f"TTS load ({engine_id})")
except Exception:
pass
if backend_cls is OmniVoiceBackend:
_model = await get_model()
else:
try:
ok, msg = backend_cls.is_available()
except Exception as exc:
ok, msg = False, f"{type(exc).__name__}: {exc}"
if not ok:
raise HTTPException(
status_code=400,
detail=f"TTS engine '{engine_id}' is not available: {_mask_hf_tokens(msg)}",
)
from api.routers.engines import _get_engine_instance
_backend = _get_engine_instance(backend_cls)
from core.device_caps import detect_host_caps
from services.engine_routing import (
routing_notice,
runtime_compute_profile_async,
)
_routing = await runtime_compute_profile_async(
backend_cls, detect_host_caps()
)
_engine_min_vram_gb = _routing["min_vram_gb"]
_routing_hardware_family = _routing.get("runtime_hardware_family")
_routing_vram_gb = _routing.get("runtime_vram_gb")
if _routing["routing_status"] == "unavailable":
raise HTTPException(status_code=400, detail=_routing["routing_reason"])
_routing_notice = routing_notice(_routing)
if _backend is not None:
from services import gpu_gateway
try:
await gpu_gateway.prewarm(
_REMOTE_OP, backend=_backend, engine=engine_id, decision=_decision,
)
except (TimeoutError, gpu_gateway.ModelLoadTimeout) as exc:
logger.warning("engine load exceeded the model-load budget: %s", exc)
raise HTTPException(
status_code=503,
detail=(
f"TTS engine '{engine_id}' did not finish loading within its "
f"model-load budget — on a first run this usually means the "
f"weight download is slow or stalled (check the engine's Weights list in Model Catalogue "
f"for progress), not that generation failed. Retry once the "
f"model shows as installed."
),
headers={"Retry-After": "30", "X-OmniVoice-Retryable": "true"},
) from exc
except HTTPException:
raise
except Exception as exc:
if type(exc).__name__ == "ModelLoadInterruptedByShutdown":
raise
from core.public_errors import model_load_failure
logger.error("engine model load failed")
return JSONResponse(
status_code=503,
content=model_load_failure(engine_id, exc),
headers={"Retry-After": "30", "X-OmniVoice-Retryable": "true"},
)
ref_audio_path = None
cleanup_ref = False
ref_lease = None
used_seed = seed
resolved_profile_id = None
language_from_profile = False
history_mode = None
stored_design_recipe = _design_recipe_json(design_recipe)
persist_ref_text_profile_id = None
request_ref_text = (ref_text or "").strip()
if profile_id:
with db_conn() as conn:
row = conn.execute("SELECT * FROM voice_profiles WHERE id=?", (profile_id,)).fetchone()
if row:
_cond = _resolve_profile_conditioning(
row, ref_text=ref_text, instruct=instruct, seed=used_seed,
language=language,
)
resolved_profile_id = None if _cond["diverged"] else profile_id
history_mode = _cond["kind"]
ref_audio_path = _cond["ref_audio_path"]
ref_text = _cond["ref_text"]
instruct = _cond["instruct"]
used_seed = _cond["seed"]
language = _cond["language"]
language_from_profile = _cond["language_from_profile"]
if _cond["persist_ref_text"]:
persist_ref_text_profile_id = profile_id
elif ref_audio is not None:
try:
suffix = _ref_upload_suffix(ref_audio.filename)
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
f.write(await ref_audio.read())
ref_audio_path = f.name
cleanup_ref = True
ref_lease = _TempReferenceLease(ref_audio_path)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
ref_picks_own_passage = False
if ref_audio_path:
from services.tts_backend import reference_duration_s
_ref_max = getattr(backend_cls, "max_ref_seconds", None)
if getattr(backend_cls, "ref_strategy", None) == "best_window" and _ref_max:
_ref_seconds = await asyncio.to_thread(reference_duration_s, ref_audio_path)
ref_picks_own_passage = _ref_seconds is not None and _ref_seconds > _ref_max
if ref_picks_own_passage and request_ref_text:
from omnivoice.utils.audio import clone_ref_transcript_too_long_message
if cleanup_ref and ref_lease is not None:
ref_lease.finish_request()
raise HTTPException(
status_code=400,
detail=clone_ref_transcript_too_long_message(_ref_seconds),
)
if ref_audio_path and not ref_text and not ref_picks_own_passage:
from services.asr_backend import transcribe_reference
try:
ref_text = await _run_with_reference_lease(
ref_lease,
lambda release: run_on_gpu_pool_guarded(
functools.partial(transcribe_reference, ref_audio_path),
what="Reference transcribe",
timeout=_generate_timeout_s(
"", execution_device=_routing["effective_device"]
),
on_abandon=release,
)
)
except TimeoutError as e:
logger.warning("reference transcribe hung (%s); using model ASR fallback", e)
ref_text = None
if ref_text and persist_ref_text_profile_id:
_persist_profile_ref_text(persist_ref_text_profile_id, ref_text)
if used_seed is None:
used_seed = random.randint(0, 2**31 - 1)
if isinstance(language, str) and language.strip().lower() == "auto":
language = None
from services.text_normalization import normalize_for_tts
text = normalize_for_tts(text, language)
if pronounce and pronunciation_enabled():
from services.pronunciation import apply_pronunciation, load_entries_from_db
try:
_pron_rows = load_entries_from_db()
except Exception:
_pron_rows = []
text = apply_pronunciation(text, _pron_rows, language)
else:
from services.pronunciation import apply_inline_overrides
text = apply_inline_overrides(text)
start_time = time.time()
from services import gpu_gateway
from services.watermark import is_enabled as _watermark_enabled
_remote_params = {
"text": text,
"language": None if (language and language.lower() == "auto") else language,
"ref_audio": ref_audio_path,
"ref_text": ref_text,
"instruct": instruct,
"duration": duration,
"speed": speed,
"num_step": num_step,
"guidance_scale": guidance_scale,
"denoise": denoise,
"postprocess_output": postprocess_output,
"t_shift": t_shift,
"layer_penalty_factor": layer_penalty_factor,
"position_temperature": position_temperature,
"class_temperature": class_temperature,
"seed": used_seed,
"max_chunk_chars": max_chunk_chars,
"crossfade_ms": crossfade_ms,
"effect_preset": effect_preset,
"wav_bits": int(wav_bits),
"watermark": bool(_watermark_enabled()),
}
_remote_call = gpu_gateway.RemoteCall(
engine=engine_id, operation=_REMOTE_OP, params=_remote_params,
)
async def _render_on_worker(on_state=None):
"""One whole render on the chosen worker → ``(tensor, sample_rate)``.
The audio comes back already effect-chained and provenance-marked: the
worker mirrors the local order (split → generate → concat → effects →
mark) so a remote take and a local take of the same request differ
only in which GPU produced them.
"""
waveform, sample_rate = await gpu_gateway.run(
_REMOTE_OP,
local=gpu_gateway.LocalCall(
_remote_only_local_call(_target_label),
what="TTS generate",
timeout=_generate_timeout_s(
text,
engine=_backend,
execution_device=_routing["effective_device"],
min_vram_gb=_engine_min_vram_gb,
hardware_family=_routing_hardware_family,
vram_gb=_routing_vram_gb,
),
min_vram_gb=_engine_min_vram_gb,
),
remote=_remote_call,
decision=_decision,
on_state=on_state,
)
if getattr(waveform, "ndim", 2) == 1:
waveform = waveform.unsqueeze(0)
return waveform, sample_rate
if stream and _remote:
_remote_headers = _apply_routing_headers(
{"X-Seed": str(used_seed) if used_seed is not None else "",
"Cache-Control": "no-cache"},
None, _decision,
)
_progress_q: asyncio.Queue = asyncio.Queue()
def _push_progress(event):
try:
_progress_q.put_nowait(dict(event or {}))
except Exception:
logger.debug("dropped a remote progress frame", exc_info=True)
async def _remote_stream_events():
import json
def _line(obj) -> bytes:
return (json.dumps(obj, separators=(",", ":")) + "\n").encode("utf-8")
render = asyncio.ensure_future(_render_on_worker(_push_progress))
try:
while not render.done():
getter = asyncio.ensure_future(_progress_q.get())
done, _pending = await asyncio.wait(
{render, getter}, return_when=asyncio.FIRST_COMPLETED
)
if getter in done:
yield _line(_remote_progress_frame(getter.result(), _target_label))
continue
getter.cancel()
while not _progress_q.empty():
yield _line(_remote_progress_frame(_progress_q.get_nowait(),
_target_label))
audio_tensor, sample_rate = await render
yield _line({
"type": "start", "sample_rate": sample_rate, "channels": 1,
"format": "pcm16", "total_chunks": 1, "crossfade_ms": 0,
"seed": used_seed,
})
yield _line({"type": "chunk", "seq": 0, "pcm": _pcm16_b64(audio_tensor)})
_, meta = await _finalize_generation(
audio_tensor, sample_rate, text=text, history_mode=history_mode,
ref_audio_path=ref_audio_path, language=language,
instruct=instruct, resolved_profile_id=resolved_profile_id,
used_seed=used_seed, start_time=start_time, already_marked=True, wav_bits=int(wav_bits),
design_recipe=stored_design_recipe,
)
yield _line({
"type": "done", "id": meta["id"], "audio_path": meta["filename"],
"duration": meta["duration"], "gen_time": meta["gen_time"],
"seed": used_seed, "sample_rate": sample_rate,
})
except (asyncio.CancelledError, GeneratorExit):
render.cancel()
raise
except ValueError as e:
logger.error("Remote generation request rejected")
from core.public_errors import stream_failure
failure = (
_language_rejection_payload(e, language, from_profile=language_from_profile)
if _is_language_rejection(str(e)) else stream_failure("invalid_request")
)
yield _line({"type": "error", **failure})
except gpu_gateway.ModelNotDownloaded as e:
logger.warning("Remote model missing on %s", _target_label)
from core.public_errors import stream_failure
yield _line({
"type": "error",
**stream_failure("model_not_downloaded"),
"engine": e.engine,
"repo_ids": e.repo_ids,
"target": e.target,
"target_label": e.target_label,
"downloadable": e.downloadable,
})
except gpu_gateway.RemoteJobFailed as e:
logger.error("Remote generate failed on %s", _target_label)
from core.public_errors import stream_failure
if _is_language_rejection(str(e)):
yield _line({"type": "error", **_language_rejection_payload(
e, language, from_profile=language_from_profile,
)})
else:
yield _line({
"type": "error", **stream_failure("generation_failed"),
"retryable": True, "target_label": e.worker_label or _target_label,
"hint": e.hint,
})
except Exception as exc:
logger.error(
"Remote generation failed (class=%s)",
type(exc).__name__,
)
from core.public_errors import stream_generation_failure
from core import error_journal
error_journal.record(
exc, route="/generate", trace=traceback.format_exc()
)
yield _line({"type": "error", **stream_generation_failure(exc)})
finally:
if not render.done():
render.cancel()
if cleanup_ref and ref_lease is not None:
ref_lease.finish_request()
return StreamingResponse(
_remote_stream_events(),
media_type="application/x-ndjson",
headers=_remote_headers,
)
if stream:
from omnivoice.utils.text import parse_pause_markers
from services.chunked_tts import split_text_into_chunks
_segments = parse_pause_markers(text)
_has_pause = len(_segments) > 1 or (_segments and _segments[0][1] > 0)
_text_chunks = [] if _has_pause else split_text_into_chunks(text, max_chunk_chars)
_dropped_sink: list = []
def _render_stream_chunk(i: int, chunk_text: str):
"""One text chunk → (raw engine tensor, preview-DSP tensor, sr).
Runs on the GPU pool. Mirrors ONE iteration of the multi-chunk
loop in _run_inference/_run_backend_inference exactly (per-chunk
deterministic seed, same generate kwargs), so concatenating the
raw parts afterwards reproduces the non-streaming output. The
preview copy gets the same effect chain the final file will get,
so what the user hears mid-stream matches the saved take.
"""
import torch
try:
if used_seed is not None:
torch.manual_seed(used_seed + i)
if _backend is not None:
_lang = None if (language and language.lower() == "auto") else language
raw = trace_call("synthesis", _backend.generate,
chunk_text, duration=None, language=_lang,
ref_audio=ref_audio_path, ref_text=ref_text,
instruct=instruct, num_step=num_step,
guidance_scale=guidance_scale, speed=speed,
denoise=denoise, postprocess_output=postprocess_output,
**({
key: value for key, value in {
"t_shift": t_shift,
"layer_penalty_factor": layer_penalty_factor,
"position_temperature": position_temperature,
"class_temperature": class_temperature,
}.items() if value is not None
} if getattr(
_backend, "supports_native_omnivoice_controls", False
) else {}),
**({"seed": used_seed + i} if used_seed is not None and (
getattr(_backend, "supports_native_omnivoice_controls", False)
or getattr(_backend, "supports_generation_seed", False)
) else {}),
)
sr = _backend.sample_rate
skip = getattr(_backend, "applies_own_mastering", False)
else:
kwargs = {}
if t_shift is not None: kwargs["t_shift"] = t_shift
if layer_penalty_factor is not None: kwargs["layer_penalty_factor"] = layer_penalty_factor
if position_temperature is not None: kwargs["position_temperature"] = position_temperature
if class_temperature is not None: kwargs["class_temperature"] = class_temperature
from services.tts_backend import generate_with_cached_ref
raw = trace_call("synthesis", generate_with_cached_ref,
_model, ref_audio=ref_audio_path, ref_text=ref_text,
text=chunk_text, language=language, instruct=instruct,
duration=None, num_step=num_step,
guidance_scale=guidance_scale, speed=speed, denoise=denoise,
postprocess_output=postprocess_output, **kwargs
)[0]
sr = _model.sampling_rate if hasattr(_model, "sampling_rate") else 24000
skip = False
preview = _apply_effect_chain(raw, sr, effect_preset, skip_mastering=skip)
return raw, preview, sr
except ValueError:
raise
except Exception as e:
_oom_friendly_reraise(e)
def _assemble_stream_chunks(parts, sr):
"""Concat + whole-take effect chain — the same tail the
non-streaming multi-chunk loop runs, as one pool job."""
from services.chunked_tts import concatenate_audio_chunks
try:
audio_out = concatenate_audio_chunks(parts, sr, crossfade_ms,
texts=_text_chunks,
sink=_dropped_sink)
skip = (getattr(_backend, "applies_own_mastering", False)
if _backend is not None else False)
return _apply_effect_chain(audio_out, sr, effect_preset, skip_mastering=skip)
except ValueError:
raise
except Exception as e:
_oom_friendly_reraise(e)
async def _stream_events():
import json
def _line(obj) -> bytes:
return (json.dumps(obj, separators=(",", ":")) + "\n").encode("utf-8")
try:
if _has_pause or len(_text_chunks) <= 1:
if _backend is not None:
audio_tensor = await _run_with_reference_lease(
ref_lease,
lambda release: run_on_gpu_pool_guarded(
functools.partial(
_run_backend_inference,
_backend, text, language, ref_audio_path, ref_text,
instruct, duration, num_step, guidance_scale, speed,
denoise, postprocess_output, used_seed, effect_preset,
max_chunk_chars, crossfade_ms, t_shift=t_shift,
layer_penalty_factor=layer_penalty_factor,
position_temperature=position_temperature,
class_temperature=class_temperature,
dropped_sink=_dropped_sink,
),
what="TTS generate",
min_vram_gb=_engine_min_vram_gb,
timeout=_generate_timeout_s(
text,
engine=_backend,
execution_device=_routing["effective_device"],
min_vram_gb=_engine_min_vram_gb,
hardware_family=_routing_hardware_family,
vram_gb=_routing_vram_gb,
),
on_abandon=release,
)
)
sample_rate = _backend.sample_rate
else:
audio_tensor = await _run_with_reference_lease(
ref_lease,
lambda release: run_on_gpu_pool_guarded(
functools.partial(
_run_inference,
_model, text, language, ref_audio_path, ref_text,
instruct, duration, num_step, guidance_scale, speed,
t_shift, denoise, postprocess_output,
layer_penalty_factor, position_temperature,
class_temperature, used_seed, effect_preset,
max_chunk_chars, crossfade_ms, dropped_sink=_dropped_sink,
),
what="TTS generate",
min_vram_gb=_engine_min_vram_gb,
timeout=_generate_timeout_s(
text,
engine=_backend,
execution_device=_routing["effective_device"],
min_vram_gb=_engine_min_vram_gb,
hardware_family=_routing_hardware_family,
vram_gb=_routing_vram_gb,
),
on_abandon=release,
)
)
sample_rate = _model.sampling_rate
yield _line({
"type": "start", "sample_rate": sample_rate, "channels": 1,
"format": "pcm16", "total_chunks": 1, "crossfade_ms": 0,
"seed": used_seed,
})
from services.watermark import mark_synthetic_async
_preview = await mark_synthetic_async(
audio_tensor, sample_rate,
context="generate.stream_preview",
)
yield _line({"type": "chunk", "seq": 0, "pcm": _pcm16_b64(_preview)})
else:
parts = []
sample_rate = None
for i, chunk_text in enumerate(_text_chunks):
raw, preview, sample_rate = await _run_with_reference_lease(
ref_lease,
lambda release: run_on_gpu_pool_guarded(
functools.partial(_render_stream_chunk, i, chunk_text),
what="TTS generate",
min_vram_gb=_engine_min_vram_gb,
timeout=_generate_timeout_s(
chunk_text,
engine=_backend,
execution_device=_routing["effective_device"],
min_vram_gb=_engine_min_vram_gb,
hardware_family=_routing_hardware_family,
vram_gb=_routing_vram_gb,
),
on_abandon=release,
)
)
parts.append(raw)
from services.watermark import mark_synthetic_async
preview = await mark_synthetic_async(
preview, sample_rate,
context="generate.stream_preview",
)
if i == 0:
yield _line({
"type": "start", "sample_rate": sample_rate,
"channels": 1, "format": "pcm16",
"total_chunks": len(_text_chunks),
"crossfade_ms": crossfade_ms, "seed": used_seed,
})
yield _line({"type": "chunk", "seq": i, "pcm": _pcm16_b64(preview)})
audio_tensor = await run_on_gpu_pool_guarded(
functools.partial(_assemble_stream_chunks, parts, sample_rate),
what="TTS assemble",
timeout=_generate_timeout_s(text, execution_device=_routing["effective_device"]),
)
_, meta = await _finalize_generation(
audio_tensor, sample_rate, text=text, history_mode=history_mode,
ref_audio_path=ref_audio_path, language=language,
instruct=instruct, resolved_profile_id=resolved_profile_id,
used_seed=used_seed, start_time=start_time, wav_bits=int(wav_bits),
design_recipe=stored_design_recipe,
)
if _dropped_sink:
yield _line({
"type": "warning", "code": "dropped_chunks",
"count": len(_dropped_sink),
"text": [t for t in _dropped_sink if t],
})
yield _line({
"type": "done", "id": meta["id"], "audio_path": meta["filename"],
"duration": meta["duration"], "gen_time": meta["gen_time"],
"seed": used_seed, "sample_rate": sample_rate,
"dropped_chunks": len(_dropped_sink),
})
except (asyncio.CancelledError, GeneratorExit):
raise
except GpuPoolBusyError as e:
logger.error("Streaming generation capacity unavailable")
from core.public_errors import stream_failure
failure = stream_failure("generation_busy")
failure["retry_after"] = getattr(e, "retry_after", 30)
yield _line({"type": "error", **failure})
except GpuJobTimeoutError:
logger.error("Streaming generation exceeded its compute budget")
from core.public_errors import stream_failure
failure = stream_failure(
"generation_timeout",
device=_routing.get("effective_device"),
)
failure["retry_after"] = 30
yield _line({"type": "error", **failure})
except ValueError as e:
logger.error("Streaming generation request rejected")
from core.public_errors import stream_failure
failure = (
_language_rejection_payload(e, language, from_profile=language_from_profile)
if _is_language_rejection(str(e)) else stream_failure("invalid_request")
)
yield _line({"type": "error", **failure})
except Exception as exc:
logger.error(
"Streaming generation failed unexpectedly (class=%s)",
type(exc).__name__,
)
from core.public_errors import stream_generation_failure
from core import error_journal
error_journal.record(
exc, route="/generate", trace=traceback.format_exc()
)
yield _line({"type": "error", **stream_generation_failure(exc)})
finally:
if cleanup_ref and ref_lease is not None:
ref_lease.finish_request()
_stream_headers = _apply_routing_headers({
"X-Seed": str(used_seed) if used_seed is not None else "",
"Cache-Control": "no-cache",
}, _routing_notice, _decision)
return StreamingResponse(
_stream_events(),
media_type="application/x-ndjson",
headers=_stream_headers,
)
_dropped_text: list = []
_already_marked = False
try:
if _remote:
audio_tensor, sample_rate = await _render_on_worker()
_already_marked = True
else:
if _backend is not None:
_local_render = functools.partial(
_run_backend_inference,
_backend, text, language, ref_audio_path, ref_text, instruct,
duration, num_step, guidance_scale, speed, denoise,
postprocess_output, used_seed, effect_preset,
max_chunk_chars, crossfade_ms, t_shift=t_shift,
layer_penalty_factor=layer_penalty_factor,
position_temperature=position_temperature,
class_temperature=class_temperature,
dropped_sink=_dropped_text,
)
else:
_local_render = functools.partial(
_run_inference,
_model, text, language, ref_audio_path, ref_text, instruct, duration,
num_step, guidance_scale, speed, t_shift, denoise,
postprocess_output, layer_penalty_factor, position_temperature,
class_temperature, used_seed, effect_preset,
max_chunk_chars, crossfade_ms, dropped_sink=_dropped_text,
)
audio_tensor = await _run_with_reference_lease(
ref_lease,
lambda release: gpu_gateway.run(
_REMOTE_OP,
local=gpu_gateway.LocalCall(
_local_render, what="TTS generate",
timeout=_generate_timeout_s(
text,
engine=_backend,
execution_device=_routing["effective_device"],
min_vram_gb=_engine_min_vram_gb,
hardware_family=_routing_hardware_family,
vram_gb=_routing_vram_gb,
),
min_vram_gb=_engine_min_vram_gb,
on_abandon=release,
),
decision=_decision,
)
)
sample_rate = (_backend.sample_rate if _backend is not None
else _model.sampling_rate)
audio_tensor, _meta = await _finalize_generation(
audio_tensor, sample_rate, text=text, history_mode=history_mode,
ref_audio_path=ref_audio_path, language=language, instruct=instruct,
resolved_profile_id=resolved_profile_id, used_seed=used_seed,
start_time=start_time, already_marked=_already_marked, wav_bits=int(wav_bits), include_wav_bytes=True,
design_recipe=stored_design_recipe,
)
audio_id = _meta["id"]
audio_filename = _meta["filename"]
audio_dur = _meta["duration"]
gen_time = _meta["gen_time"]
wav_bytes = _meta.pop("_wav_bytes")
async def _stream_wav():
chunk_size = 16384
for i in range(0, len(wav_bytes), chunk_size):
yield wav_bytes[i:i + chunk_size]
_resp_headers = {
"X-Audio-Id": audio_id,
"X-Gen-Time": str(gen_time),
"X-Audio-Path": audio_filename,
"X-Seed": str(used_seed) if used_seed is not None else "",
"X-Audio-Duration": str(audio_dur),
"Content-Length": str(len(wav_bytes)),
}
if _dropped_text:
from services.engine_routing import header_safe_reason
_resp_headers["X-OmniVoice-Dropped-Chunks"] = str(len(_dropped_text))
_lost = header_safe_reason(" | ".join(t for t in _dropped_text if t))
if _lost:
_resp_headers["X-OmniVoice-Dropped-Text"] = _lost
_apply_routing_headers(_resp_headers, _routing_notice, _decision)
return StreamingResponse(
_stream_wav(),
media_type="audio/wav",
headers=_resp_headers,
)
except HTTPException:
raise
except gpu_gateway.ModelNotDownloaded as e:
size_bytes = None
try:
from api.routers.setup.models import KNOWN_MODELS
sizes = [m.get("size_gb") for m in KNOWN_MODELS if m.get("repo_id") in e.repo_ids]
if sizes and all(size is not None for size in sizes):
size_bytes = int(sum(float(size) for size in sizes) * 1024**3)
except Exception:
pass
raise HTTPException(status_code=409, detail={
"error": "model_not_downloaded",
"message": str(e),
"engine": e.engine,
"repo_ids": e.repo_ids,
"size_bytes": size_bytes,
"target": e.target,
"target_label": e.target_label,
"downloadable": e.downloadable,
}) from e
except gpu_gateway.RemoteJobFailed as e:
logger.error("Remote generate failed on %s: %s", _target_label, e)
if _is_language_rejection(str(e)):
raise _language_rejection_http_error(
e, language, from_profile=language_from_profile
) from e
raise HTTPException(
status_code=503,
detail=f"{e} {e.hint or 'Run it on this machine instead, or pick another GPU.'}",
headers={"X-OmniVoice-Retryable": "true",
"X-OmniVoice-Routing": "remote_failed",
"Retry-After": "10"},
) from e
except GpuPoolBusyError as e:
logger.warning("Generate refused — GPU pool saturated: %s", e)
raise HTTPException(
status_code=503, detail=str(e),
headers={"Retry-After": str(e.retry_after),
"X-OmniVoice-Retryable": "true"},
) from e
except GpuJobTimeoutError as e:
logger.error("Generate timed out: %s", e)
raise HTTPException(
status_code=503, detail=str(e),
headers={"Retry-After": "30", "X-OmniVoice-Retryable": "true"},
) from e
except InvalidBinaryError as e:
logger.error("Engine binary preflight failed: %s", e)
raise HTTPException(status_code=503, detail=str(e)) from e
except ValueError as e:
logger.error("Validation failed: %s", e)
if language_from_profile and _is_language_rejection(str(e)):
raise _language_rejection_http_error(
e, language, from_profile=True
) from e
from core.failure import classify, public_hint_for_topic
_topic = classify(str(e))
_owned = public_hint_for_topic(_topic) if _topic else ""
raise HTTPException(status_code=400, detail=_owned or str(e)) from e
except Exception as e:
tb = traceback.format_exc()
logger.error("Inference failed: %s\n%s", e, tb)
raise HTTPException(
status_code=500,
detail=(
f"Couldn't synthesize audio. See Settings → Logs → Backend for the full trace. "
f"Underlying error: {_safe_exc_text(e)}"
),
)
finally:
if cleanup_ref and ref_lease is not None:
ref_lease.finish_request()
def _safe_output_path(name):
if not name:
return None
base = os.path.basename(name)
if base != name:
return None
outputs_real = os.path.realpath(OUTPUTS_DIR)
candidate = os.path.realpath(os.path.join(OUTPUTS_DIR, base))
if not candidate.startswith(outputs_real + os.sep):
return None
return candidate
def _remove_deferred_wavs(paths) -> None:
"""Delete WAVs queued by ``_remove_wav_if_unreferenced(..., defer=...)``.
Call this AFTER the ``db_conn()`` block exits (its commit succeeded): a
rolled-back delete must never have already unlinked the audio its
surviving rows still point at."""
for p in paths:
with contextlib.suppress(OSError):
os.remove(p)
def _remove_wav_if_unreferenced(conn, audio_path, exclude_ids=(), defer=None):
"""Delete a history WAV from OUTPUTS_DIR — but only when no *other*
generation_history row still references the same file.
With ``defer`` (a list) the path is appended instead of removed, so the
caller can unlink after its transaction commits (``_remove_deferred_wavs``).
History WAVs are uniquely owned by their row (lock/save-as-profile COPY
into VOICES_DIR, exports copy to the user's destination), so this guard is
normally a no-op — it exists so any future path that duplicates a row can
never make a delete/prune yank audio out from under a surviving take."""
if not audio_path:
return
p = _safe_output_path(audio_path)
if not p or not os.path.exists(p):
return
placeholders = ",".join("?" for _ in exclude_ids)
others = conn.execute(
"SELECT COUNT(*) FROM generation_history WHERE audio_path=?"
+ (f" AND id NOT IN ({placeholders})" if exclude_ids else ""),
(audio_path, *exclude_ids),
).fetchone()[0]
if others:
return
if defer is not None:
defer.append(p)
return
with contextlib.suppress(OSError):
os.remove(p)
def pronunciation_enabled() -> bool:
"""The pronunciation dictionary is on unless OMNIVOICE_PRONUNCIATION is
"0"/"false"/"no"/"off". Env only: a prefs key was once read here too, but
no Settings surface ever wrote it."""
value = os.environ.get("OMNIVOICE_PRONUNCIATION")
return value is None or value.strip().lower() not in ("0", "false", "no", "off", "")
HISTORY_CAP_PREF_KEY = "generation_history_cap"
DEFAULT_HISTORY_CAP = 200
def _history_cap() -> int:
from core import prefs
try:
cap = int(prefs.get(HISTORY_CAP_PREF_KEY, DEFAULT_HISTORY_CAP))
except (TypeError, ValueError):
return DEFAULT_HISTORY_CAP
return max(0, cap)
def _prune_history_over_cap(*, keep_id: str | None = None) -> int:
"""Retention: keep the newest ``_history_cap()`` takes; delete the oldest
UNstarred rows over the cap plus their WAVs (via the unreferenced guard).
Starred takes are never pruned — even when they alone exceed the cap.
The currently generated take is retained until a later generation.
Returns the number of rows pruned."""
cap = _history_cap()
if cap <= 0:
return 0
doomed: list[str] = []
with db_conn() as conn:
total = conn.execute("SELECT COUNT(*) FROM generation_history").fetchone()[0]
excess = total - cap
if excess <= 0:
return 0
victims = conn.execute(
"SELECT id, audio_path FROM generation_history "
"WHERE COALESCE(starred, 0)=0 AND (? IS NULL OR id != ?) "
"ORDER BY created_at ASC LIMIT ?",
(keep_id, keep_id, excess),
).fetchall()
if not victims:
return 0
victim_ids = [r["id"] for r in victims]
conn.executemany(
"DELETE FROM generation_history WHERE id=?", [(i,) for i in victim_ids]
)
for r in victims:
_remove_wav_if_unreferenced(
conn, r["audio_path"], exclude_ids=victim_ids, defer=doomed
)
_remove_deferred_wavs(doomed)
logger.info("history retention: pruned %d takes over the %d cap", len(victims), cap)
return len(victims)
@router.get("/history")
def list_history():
"""The newest 50 generations plus every starred take, newest first, kept to
rows whose audio still exists on disk.
Starred takes ride along past the 50-row window so a keeper can never age
off the rail. Rows whose WAV was deleted out-of-band (cleared outputs dir,
manual cleanup) used to come back anyway and render dead players that 404
on every fetch; prune them here so the UI never sees them again."""
query = (
"SELECT * FROM generation_history WHERE COALESCE(starred, 0)=1 "
"OR id IN (SELECT id FROM generation_history ORDER BY created_at DESC LIMIT 50) "
"ORDER BY created_at DESC"
)
with db_conn() as conn:
try:
rows = conn.execute(query).fetchall()
except sqlite3.OperationalError:
ensure_schema()
rows = conn.execute(query).fetchall()
alive, stale_ids = [], []
for r in rows:
p = _safe_output_path(r["audio_path"]) if r["audio_path"] else None
if r["audio_path"] and (not p or not os.path.exists(p)):
stale_ids.append(r["id"])
else:
alive.append(dict(r))
if stale_ids:
conn.executemany(
"DELETE FROM generation_history WHERE id=?",
[(i,) for i in stale_ids],
)
logger.info("pruned %d stale history rows (audio file gone)", len(stale_ids))
return alive
class _StarBody(BaseModel):
starred: bool
@router.put("/history/{history_id}/starred")
def set_history_starred(history_id: str, body: _StarBody):
"""Star/unstar a take. Starred takes survive the retention cap and always
appear in GET /history regardless of the recency window."""
def _update():
with db_conn() as conn:
cur = conn.execute(
"UPDATE generation_history SET starred=? WHERE id=?",
(1 if body.starred else 0, history_id),
)
return cur.rowcount
try:
changed = _update()
except sqlite3.OperationalError as e:
logger.warning("star update failed (%s); healing schema + retrying", e)
ensure_schema()
changed = _update()
if not changed:
raise HTTPException(
status_code=404,
detail="That take no longer exists — it may have been pruned or deleted.",
)
event_bus.emit("generation_history", {"action": "starred", "id": history_id})
return {"id": history_id, "starred": body.starred}
@router.delete("/history")
def clear_history():
with db_conn() as conn:
rows = conn.execute("SELECT audio_path FROM generation_history").fetchall()
conn.execute("DELETE FROM generation_history")
for r in rows:
p = _safe_output_path(r["audio_path"])
if p and os.path.exists(p):
with contextlib.suppress(OSError):
os.remove(p)
event_bus.emit("generation_history")
return {"cleared": True}
@router.delete("/history/{history_id}")
def delete_single_history(history_id: str):
doomed: list[str] = []
with db_conn() as conn:
row = conn.execute("SELECT audio_path FROM generation_history WHERE id=?", (history_id,)).fetchone()
conn.execute("DELETE FROM generation_history WHERE id=?", (history_id,))
if row:
_remove_wav_if_unreferenced(
conn, row["audio_path"], exclude_ids=(history_id,), defer=doomed
)
_remove_deferred_wavs(doomed)
event_bus.emit("generation_history", {"action": "deleted", "id": history_id})
return {"deleted": True}