feat: add background voice design jobs
This commit is contained in:
parent
f14e1d2e92
commit
a53f65aa05
@ -29,6 +29,8 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
|
||||
batch work is capped at two concurrent operations.
|
||||
- **Optional Background Transcription.** `/api/transcribe` now accepts `background: true` and exposes bounded job
|
||||
status/cancellation endpoints with timeout and failure states; existing synchronous callers remain compatible.
|
||||
- **Optional Background Voice Design.** `/api/voice-design` now accepts `background: true` and exposes bounded job
|
||||
status/cancellation endpoints with queue state, timeout and failure handling; existing synchronous callers remain compatible.
|
||||
- **STT Cancellation Safety.** Queued transcription cancels before consuming the single STT slot, and timed-out
|
||||
requests keep the device slot held until the underlying worker actually exits.
|
||||
- **Credential Boundary Review.** Provider credentials are now documented as server-only; only the explicitly
|
||||
|
||||
4
TODO.md
4
TODO.md
@ -63,8 +63,8 @@ the first block before investing in architecture or polish.
|
||||
a 4 GB VM/container, and measure browser, Docker and temporary-disk usage. Automated browser emulation now proves
|
||||
the 4 GB shared-pool cap; physical VM/container and resource measurements remain.
|
||||
- [ ] **L — Add background jobs and bounded queues.** Image generation already has the full interactive job path,
|
||||
and transcription now has an optional ID/status/cancel/timeout path while preserving synchronous callers. Voice
|
||||
design and casting still need migration to the same user-facing durable job model, plus retries and persistence.
|
||||
and transcription plus voice design now have optional ID/status/cancel/timeout paths while preserving synchronous
|
||||
callers. Casting still needs migration to the same user-facing durable job model, plus retries and persistence.
|
||||
- [ ] **L — Run the Book 2 generalization regression.** Confirm speaker-attribution and
|
||||
confidence rules are not overfit to *Der Sturm*.
|
||||
- [x] **L — Add character forms inside one character record.** `sheet.forms[]` stores bounded alternate appearances,
|
||||
|
||||
151
routes/tts.py
151
routes/tts.py
@ -58,6 +58,9 @@ _SEED_FINDER_VOICE_LOCK = threading.Lock()
|
||||
_TTS_GENERATION_SLOTS = asyncio.Semaphore(2)
|
||||
_VOICE_DESIGN_IDEMPOTENCY: dict[str, tuple[str, dict, float]] = {}
|
||||
_VOICE_DESIGN_IDEMPOTENCY_LOCK = threading.Lock()
|
||||
_VOICE_DESIGN_JOBS: dict[str, dict] = {}
|
||||
_VOICE_DESIGN_JOB_MAX = 128
|
||||
_VOICE_DESIGN_JOB_TIMEOUT = 15 * 60
|
||||
|
||||
|
||||
async def _bounded_tts_call(fn, *args):
|
||||
@ -71,6 +74,88 @@ async def _bounded_tts_call(fn, *args):
|
||||
finally:
|
||||
_TTS_GENERATION_SLOTS.release()
|
||||
|
||||
|
||||
def _voice_design_job_set(job_id: str, **fields) -> None:
|
||||
job = _VOICE_DESIGN_JOBS.get(job_id)
|
||||
if job is not None:
|
||||
job.update(fields)
|
||||
job["updated"] = time.time()
|
||||
|
||||
|
||||
def _voice_design_job_trim() -> None:
|
||||
if len(_VOICE_DESIGN_JOBS) <= _VOICE_DESIGN_JOB_MAX:
|
||||
return
|
||||
for job_id, job in sorted(_VOICE_DESIGN_JOBS.items(), key=lambda item: item[1].get("updated", 0)):
|
||||
if job.get("done"):
|
||||
_VOICE_DESIGN_JOBS.pop(job_id, None)
|
||||
if len(_VOICE_DESIGN_JOBS) <= _VOICE_DESIGN_JOB_MAX:
|
||||
break
|
||||
|
||||
|
||||
async def _voice_design_execute(data: dict, idempotency_key: str, request_fingerprint: str) -> dict:
|
||||
instruct: str = data.get("instruct", "").strip()
|
||||
sample_text: str = data.get("sample_text", "Hello! This is a voice design sample.").strip()
|
||||
language: str = data.get("language", "Auto")
|
||||
gender: str = data.get("gender", "")
|
||||
dialogue = bool(data.get("dialogue"))
|
||||
if not instruct:
|
||||
raise HTTPException(400, "instruct (voice description) is required")
|
||||
|
||||
settings = _load_settings()
|
||||
try:
|
||||
if dialogue:
|
||||
parsed = _parse_voice_design_dialogue(instruct, sample_text)
|
||||
if not parsed:
|
||||
raise RuntimeError("dialogue mode needs speaker profiles and Speaker: text turns")
|
||||
audio, _media_type = await _bounded_tts_call(
|
||||
_voice_design_request_audio, instruct, sample_text, language, settings, "",
|
||||
)
|
||||
else:
|
||||
audio, _media_type = await _bounded_tts_call(
|
||||
_voice_design_request_audio, instruct, sample_text, language, settings, gender,
|
||||
)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(502, f"Voice design error: {e}")
|
||||
|
||||
audio, _media_type, _ext, _duration_sec, _clipped = _prepare_proxy_audio(audio, "audio/wav", "wav")
|
||||
tmp = TEMP_DIR / f"{uuid.uuid4().hex}_designed.wav"
|
||||
tmp.write_bytes(audio)
|
||||
fid = uuid.uuid4().hex
|
||||
_registry_put(fid, tmp)
|
||||
result = {"id": fid, "duration": _duration(tmp)}
|
||||
if idempotency_key:
|
||||
with _VOICE_DESIGN_IDEMPOTENCY_LOCK:
|
||||
_VOICE_DESIGN_IDEMPOTENCY[idempotency_key] = (request_fingerprint, result, time.time())
|
||||
if len(_VOICE_DESIGN_IDEMPOTENCY) > 256:
|
||||
oldest = min(_VOICE_DESIGN_IDEMPOTENCY, key=lambda k: _VOICE_DESIGN_IDEMPOTENCY[k][2])
|
||||
_VOICE_DESIGN_IDEMPOTENCY.pop(oldest, None)
|
||||
return result
|
||||
|
||||
|
||||
async def _run_voice_design_job(job_id: str, data: dict, idempotency_key: str, request_fingerprint: str) -> None:
|
||||
if _VOICE_DESIGN_JOBS.get(job_id, {}).get("cancel_requested"):
|
||||
_voice_design_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True,
|
||||
error="Voice design cancelled")
|
||||
return
|
||||
_voice_design_job_set(job_id, stage="running", detail="Generating voice preview…")
|
||||
call = asyncio.create_task(_voice_design_execute(data, idempotency_key, request_fingerprint))
|
||||
try:
|
||||
result = await asyncio.wait_for(asyncio.shield(call), timeout=_VOICE_DESIGN_JOB_TIMEOUT)
|
||||
if _VOICE_DESIGN_JOBS.get(job_id, {}).get("cancel_requested"):
|
||||
_voice_design_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True,
|
||||
error="Voice design cancelled")
|
||||
else:
|
||||
_voice_design_job_set(job_id, stage="done", detail="Done", done=True, result=result)
|
||||
except asyncio.TimeoutError:
|
||||
call.add_done_callback(lambda task: task.exception() if not task.cancelled() else None)
|
||||
_voice_design_job_set(job_id, stage="error", detail="Timed out", done=True,
|
||||
error="Voice design timed out")
|
||||
except Exception as exc:
|
||||
message = str(exc.detail) if isinstance(exc, HTTPException) else str(exc)
|
||||
_voice_design_job_set(job_id, stage="error", detail="Failed", done=True, error=message)
|
||||
|
||||
# ── TTS backend helpers ───────────────────────────────────────────────────────
|
||||
|
||||
def _backend_port_label(url: str) -> str:
|
||||
@ -795,44 +880,40 @@ async def voice_design(request: Request):
|
||||
if previous[0] != request_fingerprint:
|
||||
raise HTTPException(409, "Idempotency-Key was already used for a different request")
|
||||
return {**previous[1], "deduplicated": True}
|
||||
instruct: str = data.get("instruct", "").strip()
|
||||
sample_text: str = data.get("sample_text", "Hello! This is a voice design sample.").strip()
|
||||
language: str = data.get("language", "Auto")
|
||||
gender: str = data.get("gender", "")
|
||||
dialogue = bool(data.get("dialogue"))
|
||||
|
||||
if not instruct:
|
||||
if not str(data.get("instruct") or "").strip():
|
||||
raise HTTPException(400, "instruct (voice description) is required")
|
||||
if data.get("background") is True:
|
||||
job_id = uuid.uuid4().hex
|
||||
job_data = dict(data)
|
||||
job_data.pop("background", None)
|
||||
_VOICE_DESIGN_JOBS[job_id] = {
|
||||
"job_id": job_id, "stage": "queued", "detail": "Queued", "done": False,
|
||||
"created": time.time(), "updated": time.time(), "cancel_requested": False,
|
||||
}
|
||||
_voice_design_job_trim()
|
||||
asyncio.create_task(_run_voice_design_job(job_id, job_data, idempotency_key, request_fingerprint))
|
||||
return {"job_id": job_id, "stage": "queued"}
|
||||
return await _voice_design_execute(data, idempotency_key, request_fingerprint)
|
||||
|
||||
settings = _load_settings()
|
||||
try:
|
||||
if dialogue:
|
||||
parsed = _parse_voice_design_dialogue(instruct, sample_text)
|
||||
if not parsed:
|
||||
raise RuntimeError("dialogue mode needs speaker profiles and Speaker: text turns")
|
||||
audio, _media_type = await asyncio.to_thread(
|
||||
_voice_design_request_audio, instruct, sample_text, language, settings, "",
|
||||
)
|
||||
else:
|
||||
audio, _media_type = await asyncio.to_thread(
|
||||
_voice_design_request_audio, instruct, sample_text, language, settings, gender,
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(502, f"Voice design error: {e}")
|
||||
|
||||
audio, _media_type, _ext, _duration_sec, _clipped = _prepare_proxy_audio(audio, "audio/wav", "wav")
|
||||
tmp = TEMP_DIR / f"{uuid.uuid4().hex}_designed.wav"
|
||||
tmp.write_bytes(audio)
|
||||
fid = uuid.uuid4().hex
|
||||
_registry_put(fid, tmp)
|
||||
result = {"id": fid, "duration": _duration(tmp)}
|
||||
if idempotency_key:
|
||||
with _VOICE_DESIGN_IDEMPOTENCY_LOCK:
|
||||
_VOICE_DESIGN_IDEMPOTENCY[idempotency_key] = (request_fingerprint, result, time.time())
|
||||
if len(_VOICE_DESIGN_IDEMPOTENCY) > 256:
|
||||
oldest = min(_VOICE_DESIGN_IDEMPOTENCY, key=lambda k: _VOICE_DESIGN_IDEMPOTENCY[k][2])
|
||||
_VOICE_DESIGN_IDEMPOTENCY.pop(oldest, None)
|
||||
return result
|
||||
@router.get("/api/voice-design/status/{job_id}")
|
||||
async def voice_design_status(job_id: str):
|
||||
job = _VOICE_DESIGN_JOBS.get(job_id)
|
||||
if job is None:
|
||||
raise HTTPException(404, "Voice design job not found")
|
||||
return dict(job)
|
||||
|
||||
|
||||
@router.post("/api/voice-design/cancel/{job_id}")
|
||||
async def voice_design_cancel(job_id: str):
|
||||
job = _VOICE_DESIGN_JOBS.get(job_id)
|
||||
if job is None:
|
||||
raise HTTPException(404, "Voice design job not found")
|
||||
if job.get("done"):
|
||||
return dict(job)
|
||||
job["cancel_requested"] = True
|
||||
_voice_design_job_set(job_id, detail="Cancellation requested")
|
||||
return dict(_VOICE_DESIGN_JOBS[job_id])
|
||||
|
||||
|
||||
@router.post("/api/tts-route-test")
|
||||
|
||||
@ -125,6 +125,24 @@ class SecurityValidationTests(unittest.TestCase):
|
||||
finally:
|
||||
stt._STT_JOBS.pop(job_id, None)
|
||||
|
||||
def test_voice_design_background_job_honours_queued_cancellation(self):
|
||||
import routes.tts as tts
|
||||
from unittest.mock import patch
|
||||
job_id = "voice-design-test-job"
|
||||
now = 0.0
|
||||
tts._VOICE_DESIGN_JOBS[job_id] = {
|
||||
"stage": "queued", "detail": "Queued", "created": now, "updated": now,
|
||||
"done": False, "cancel_requested": True,
|
||||
}
|
||||
try:
|
||||
with patch.object(tts, "_voice_design_execute") as execute:
|
||||
asyncio.run(tts._run_voice_design_job(job_id, {"instruct": "test"}, "", ""))
|
||||
self.assertTrue(tts._VOICE_DESIGN_JOBS[job_id]["done"])
|
||||
self.assertEqual(tts._VOICE_DESIGN_JOBS[job_id]["stage"], "cancelled")
|
||||
execute.assert_not_called()
|
||||
finally:
|
||||
tts._VOICE_DESIGN_JOBS.pop(job_id, None)
|
||||
|
||||
def test_fish_cache_eviction_is_bounded(self):
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
Loading…
Reference in New Issue
Block a user