fix(tts-server): bound admission, not generation, by the 5s deadline

TODO.md

@@ -28,11 +28,53 @@ lost silently.
   case) and logs each retry at WARNING so recovered outages stay visible. A
   4xx or the request-body 500 is not retried. The backoff sleeps are
   cancellation points, so barge-in still cuts a retrying turn.
-- [ ] **Root-cause the server-side 503.** Determine why the DGX
-  `/v1/audio/speech` route rejects a request under conversational load —
-  exclusive model lock contention, worker restart, or Traefik-level shedding.
-  The agent-side retry above treats the symptom; this is the cause. Decide
-  from the finding whether the retry needs a backoff longer than one turn.
+- [x] **Root-caused the server-side 503 (2026-08-01).** Not lock contention,
+  not a worker restart, not Traefik shedding — all three are ruled out by the
+  evidence below. The request path applies the *admission* budget to the
+  *whole* request, so a synthesis that generates too slowly is killed
+  mid-flight and reported as an admission timeout.
+
+  Mechanism, in `dgx/tts/api.py::_run_stable_synthesis`: `deadline` is set at
+  request entry (`monotonic() + DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS`,
+  **5.0 s**, `api.py:24`) and then enforced over the entire request. Line 150
+  raises `SynthesisAdmissionTimeout` once the deadline passes even though the
+  worker thread already holds the model lock and is actively generating, and
+  lines 159–161 discard a *successfully completed* result if it landed after
+  the deadline. `api.py:99` maps that to `503 "synthesis admission timed out"`.
+  The name says admission; the semantics are "total request must finish in
+  5 s". `create_app` is called without the argument (`server.py:55`), so the
+  5 s is effectively hardcoded — there is no env knob to widen it.
+
+  Evidence (DGX `gx10-d624`, all timestamps UTC):
+  - `qwen3-tts` access log: `18:28:54 200`, `18:29:08 503`, `18:29:11 200`.
+    Requests are serialized ~14 s apart, so the agent had nothing else in
+    flight; no `499` was logged between them, so the model lock was free and
+    admission was instant. The 5 s therefore elapsed inside generation.
+  - The 503 was logged by uvicorn (the app), and `proxy-traefik-1` logged zero
+    503s in the hour — so it is not proxy-level shedding.
+  - No ERROR, traceback, or restart in the container log; the container has
+    been `Up (healthy)` throughout — so it is not a worker restart.
+  - Measured idle latency inside the container (bypassing Traefik):
+    `"Ja."` 0.3–0.5 s, the sentence that failed 1.5–1.7 s, a long sentence
+    2.9–3.5 s. A normal long sentence already consumes ~70 % of the budget
+    when the GPU is otherwise idle.
+  - Contention at the moment of failure: the request started ~18:29:03.8
+    (5 s before the logged 503) while the shared GPU was busy — vLLM
+    `companion-llm` reported `Running: 1` with generation throughput at
+    18:29:05, and `qwen3-asr` had transcribed at 18:29:01.7. A ~3.3x slowdown
+    on a 1.5 s sentence is enough to cross 5 s.
+
+- [ ] **Fix the TTS admission budget (server side, `dgx/tts/`).** Separate
+  admission from execution: have `clone_runtime.synthesize` set an `admitted`
+  event once `_acquire_synthesis` returns, enforce the deadline in
+  `_run_stable_synthesis` only while `not admitted.is_set()`, and delete the
+  post-completion deadline check that throws away good audio. After admission
+  the request should be bounded by client disconnect (already detected) and
+  the client's own 30 s HTTP timeout. Make the timeout env-configurable while
+  there. Needs a GPU image rebuild + redeploy on the DGX.
+  The agent-side retry (above) already masks this: the live 503 was followed
+  by a 200 three seconds later, which is exactly what the retry now does
+  automatically. Backoff does **not** need to be longer than one turn.
 - [x] **Caller ID format confirmed live.** Lifecycle logging in
   `agent/answer_policy.py` is verified for both an internal FRITZ!Box extension
   (`**613`, 18:27 UTC) and an external mobile call (`015100000001`, 18:33 UTC).

dgx/tts/api.py

@@ -4,7 +4,7 @@
 
 import asyncio
 import io
-from collections.abc import AsyncIterator, Iterator
+from collections.abc import AsyncIterator, Iterator, Mapping
 from contextlib import suppress
 from dataclasses import dataclass
 from threading import Event, Lock
@@ -21,7 +21,25 @@
 from dgx.tts.profiles import ProfileError
 
 MAX_SPEECH_INPUT_CHARACTERS = 2_000
+# Budget for *admission* only — executor queue time plus the wait for the
+# exclusive model lock. Generation itself is not bounded by it; see
+# _run_stable_synthesis.
 DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS = 5.0
+SYNTHESIS_ADMISSION_TIMEOUT_ENV = "SYNTHESIS_ADMISSION_TIMEOUT_SECONDS"
+
+
+def admission_timeout_from_env(environ: Mapping[str, str]) -> float:
+    """Read the admission budget from the environment, falling back to default.
+
+    Deliberately forgiving: a malformed or non-positive value falls back rather
+    than refusing to start, because a TTS outage takes every call down with it.
+    """
+
+    try:
+        value = float(environ.get(SYNTHESIS_ADMISSION_TIMEOUT_ENV, ""))
+    except ValueError:
+        return DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS
+    return value if value > 0 else DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS
 
 
 @dataclass(frozen=True)
@@ -133,6 +151,12 @@ async def _run_stable_synthesis(
 ) -> tuple[list[Any], int]:
     deadline = monotonic() + admission_timeout_seconds
     cancelled = Event()
+    # Set by the runtime once it holds the model lock. The deadline bounds
+    # admission only — time spent queued in the executor or waiting for the
+    # lock, where the caller may already be gone. Enforcing it past admission
+    # killed in-flight generation and reported it as a 503, losing a whole
+    # response turn whenever the shared GPU made a sentence run long.
+    admitted = Event()
     operation = asyncio.create_task(
         asyncio.to_thread(
             runtime.synthesize,
@@ -142,21 +166,22 @@ async def _run_stable_synthesis(
             cancel_event=cancelled,
             lock_timeout=admission_timeout_seconds,
             admission_deadline=deadline,
+            admitted=admitted,
         )
     )
     try:
         while True:
             remaining = deadline - monotonic()
-            if remaining <= 0:
+            if remaining <= 0 and not admitted.is_set():
                 await _cancel_stable_operation(operation, cancelled)
                 raise SynthesisAdmissionTimeout()
             done, _pending = await asyncio.wait(
                 {operation},
-                timeout=min(0.025, remaining),
+                timeout=0.025 if admitted.is_set() else min(0.025, remaining),
             )
             if operation in done:
                 result = operation.result()
-                if monotonic() >= deadline:
+                if monotonic() >= deadline and not admitted.is_set():
                     await _cancel_stable_operation(operation, cancelled)
                     raise SynthesisAdmissionTimeout()
                 return result

dgx/tts/clone_runtime.py

@@ -58,12 +58,18 @@ def synthesize(
         cancel_event: Event | None = None,
         lock_timeout: float | None = None,
         admission_deadline: float | None = None,
+        admitted: Event | None = None,
     ) -> tuple[list[Any], int]:
         profile = resolve_profile(voice, self._profiles, self._default_profile_id)
         deadline = self._acquire_synthesis(cancel_event, lock_timeout, admission_deadline)
         try:
             if deadline is not None and monotonic() >= deadline:
                 raise SynthesisAdmissionTimeout()
+            # Admission is complete: the model lock is held and generation is
+            # about to start. The caller stops enforcing the admission deadline
+            # from here, so this must never be signalled earlier.
+            if admitted is not None:
+                admitted.set()
             try:
                 return self._model.generate_voice_clone(
                     text=text,

dgx/tts/server.py

@@ -7,7 +7,7 @@
 
 import torch
 
-from dgx.tts.api import HealthMetadata, create_app
+from dgx.tts.api import HealthMetadata, admission_timeout_from_env, create_app
 from dgx.tts.clone_runtime import CloneRuntime
 from dgx.tts.profiles import load_profiles
 from dgx.tts.runtime import require_gb10_cuda
@@ -54,7 +54,8 @@ def _build_app():
     )
     return create_app(
         runtime,
-        HealthMetadata(
+        synthesis_admission_timeout_seconds=admission_timeout_from_env(os.environ),
+        health=HealthMetadata(
             model_revision=MODEL_REVISION,
             default_profile=DEFAULT_PROFILE,
             profiles_loaded=tuple(sorted(profiles)),

tests/test_tts_api.py

@@ -18,12 +18,14 @@
 import pytest
 
 from dgx.tts.api import (
+    DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS,
     MAX_SPEECH_INPUT_CHARACTERS,
     HealthMetadata,
     SpeechRequest,
     _cancel_stable_operation,
     _ClientDisconnected,
     _run_stable_synthesis,
+    admission_timeout_from_env,
     create_app,
     encode_pcm_stream,
 )
@@ -511,7 +513,122 @@ def synthesize(self, *args, **kwargs):
         executor.shutdown(wait=True)
 
 
-async def test_stable_absolute_deadline_rejects_late_success() -> None:
+def test_admission_timeout_from_env_defaults_when_unset_or_invalid() -> None:
+    assert admission_timeout_from_env({}) == DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS
+    for bad in ("", "abc", "0", "-1"):
+        assert (
+            admission_timeout_from_env({"SYNTHESIS_ADMISSION_TIMEOUT_SECONDS": bad})
+            == DEFAULT_SYNTHESIS_ADMISSION_TIMEOUT_SECONDS
+        )
+
+
+def test_admission_timeout_from_env_reads_positive_override() -> None:
+    assert admission_timeout_from_env({"SYNTHESIS_ADMISSION_TIMEOUT_SECONDS": "12.5"}) == 12.5
+
+
+async def test_admitted_synthesis_outlives_the_admission_deadline() -> None:
+    """Generation slower than the admission budget must still be delivered.
+
+    The live 503 came from here: the deadline bounds *admission*, but it was
+    enforced over the whole request, so a sentence whose generation ran long
+    (a busy shared GPU) was killed mid-flight and reported as an admission
+    timeout. Once the worker holds the model lock, only client disconnect and
+    the client's own HTTP timeout may end the request.
+    """
+
+    class AdmittedSlowRuntime(_StableRuntime):
+        def __init__(self) -> None:
+            super().__init__()
+            self.started = threading.Event()
+            self.release = threading.Event()
+            self.finished = threading.Event()
+            self.cancel_event: threading.Event | None = None
+            self.completed_after_deadline: bool | None = None
+
+        def synthesize(
+            self,
+            text: str,
+            _voice: str | None,
+            _language: str | None,
+            *,
+            cancel_event: threading.Event,
+            **_admission: Any,
+        ) -> tuple[list[np.ndarray], int]:
+            deadline = _admission["admission_deadline"]
+            # The real CloneRuntime signals this once it holds the model lock.
+            _admission["admitted"].set()
+            self.calls.append(text)
+            self.cancel_event = cancel_event
+            self.started.set()
+            try:
+                self.release.wait(timeout=2)
+                self.completed_after_deadline = time.monotonic() >= deadline
+                return [np.array([0.25], dtype=np.float32)], 24_000
+            finally:
+                self.finished.set()
+
+    runtime = AdmittedSlowRuntime()
+    operation = asyncio.create_task(
+        _run_stable_synthesis(
+            runtime,
+            SpeechRequest(input="slow-but-admitted"),
+            _DisconnectRequest(),
+            0.10,
+        )
+    )
+    timer: threading.Timer | None = None
+    try:
+        await _wait_for_thread_event(runtime.started)
+        timer = threading.Timer(0.12, runtime.release.set)
+        timer.start()
+
+        audios, sample_rate = await operation
+
+        assert runtime.completed_after_deadline is True
+        assert sample_rate == 24_000
+        assert np.asarray(audios[0]).size == 1
+        assert runtime.cancel_event is not None
+        assert not runtime.cancel_event.is_set()
+    finally:
+        runtime.release.set()
+        if timer is not None:
+            timer.cancel()
+        if not operation.done():
+            operation.cancel()
+        with suppress(asyncio.CancelledError, Exception):
+            await operation
+        await _wait_for_thread_event(runtime.finished)
+
+
+async def test_slow_generation_returns_audio_not_503() -> None:
+    """End-to-end counterpart: the route must not turn slow audio into a 503."""
+
+    class SlowAdmittedRuntime(FakeRuntime):
+        def synthesize(
+            self,
+            text: str,
+            voice: str | None,
+            language: str | None,
+            **admission,
+        ):
+            admission["admitted"].set()
+            time.sleep(0.15)  # generation outlasts the 0.05 s admission budget
+            return [np.array([0.0, 0.5, -0.5], dtype=np.float32)], 24_000
+
+    app = create_app(SlowAdmittedRuntime(), _health(), synthesis_admission_timeout_seconds=0.05)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="http://test"
+    ) as client:
+        response = await client.post("/v1/audio/speech", json={"input": "langsam"})
+
+    assert response.status_code == 200
+    assert response.headers["content-type"] == "audio/wav"
+
+
+async def test_stable_absolute_deadline_rejects_never_admitted_late_success() -> None:
+    # A worker that never signals admission is still bounded by the deadline:
+    # it may be stuck in the executor queue with the caller long gone, so a
+    # result that arrives late is discarded and the model is cancelled.
     class LateSuccessRuntime(_StableRuntime):
         def __init__(self) -> None:
             super().__init__()

tests/test_tts_clone_runtime.py

@@ -286,6 +286,63 @@ def generate_voice_clone(self, **kwargs):
     assert model.generated_texts == ["active", "following"]
 
 
+def test_synthesize_signals_admission_before_generation(tmp_path: Path) -> None:
+    # The request path stops enforcing the admission deadline once this event
+    # is set, so it must be set only after the model lock is held and never
+    # before generation starts.
+    profile = _profile(tmp_path)
+    admitted = threading.Event()
+    admitted_at_generation: bool | None = None
+
+    class SignalCheckingModel(ExactCloneModel):
+        def generate_voice_clone(self, **kwargs):
+            nonlocal admitted_at_generation
+            admitted_at_generation = admitted.is_set()
+            return self.warm_audio, self.sample_rate
+
+    model = SignalCheckingModel(profile)
+    runtime = CloneRuntime(model, {profile.profile_id: profile}, profile.profile_id)
+
+    runtime.synthesize("hallo", profile.profile_id, "de", admitted=admitted)
+
+    assert admitted_at_generation is True
+    assert admitted.is_set()
+
+
+def test_admission_timeout_leaves_admission_unsignalled(tmp_path: Path) -> None:
+    profile = _profile(tmp_path)
+    admitted = threading.Event()
+    active_started = threading.Event()
+    release_active = threading.Event()
+
+    class BlockingSynthesisModel(ExactCloneModel):
+        def generate_voice_clone(self, **kwargs):
+            if kwargs["text"] == "active":
+                active_started.set()
+                release_active.wait(timeout=2)
+            return self.warm_audio, self.sample_rate
+
+    model = BlockingSynthesisModel(profile)
+    runtime = CloneRuntime(model, {profile.profile_id: profile}, profile.profile_id)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        active = executor.submit(runtime.synthesize, "active", profile.profile_id, "de")
+        assert active_started.wait(timeout=1)
+        timed_out = executor.submit(
+            runtime.synthesize,
+            "timed-out",
+            profile.profile_id,
+            "de",
+            lock_timeout=0.05,
+            admitted=admitted,
+        )
+        with pytest.raises(SynthesisAdmissionTimeout):
+            timed_out.result(timeout=1)
+        assert not admitted.is_set()
+        release_active.set()
+        active.result(timeout=1)
+
+
 def test_clone_runtime_holds_lock_for_complete_stream_lifetime(tmp_path: Path) -> None:
     profile = _profile(tmp_path)
     first_chunk = threading.Event()