From beb95108ffdc5df23d65fb58a6462eb38c40aeb2 Mon Sep 17 00:00:00 2001 From: berges99 Date: Wed, 29 Jul 2026 19:02:36 +0200 Subject: [PATCH 1/5] feat(server): single-session lifetime and relay-only ICE for serverless voice boxes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TIMBAL_VOICE_SINGLE_SESSION=1 makes the server serve exactly one voice session (WS or RTC) and exit 0 — after draining the platform recording upload, since the process is the only thing holding that data. Boxes nobody connects to exit after TIMBAL_VOICE_IDLE_EXIT_SECS (default 60, boot → media established); a second session gets 409 (RTC) / 1008 (WS). Needed on serverless workforce deployments where the platform can't see the call (media flows browser ↔ TURN ↔ box) and reaps on process exit. TIMBAL_VOICE_RTC_FORCE_RELAY=1 restricts ICE to the TURN relay: skip STUN, allocate on TIMBAL_TURN_URL, and strip host/srflx candidates from the answer SDP (aiortc has no iceTransportPolicy) — private-subnet boxes advertise only reachable candidates. Degrades loudly to all candidates when TURN is unconfigured or the allocation yields no relay candidate. --- python/tests/server/test_rtc_force_relay.py | 86 +++++++++ python/tests/server/test_single_session.py | 188 ++++++++++++++++++++ python/tests/server/test_voice_rtc.py | 75 ++++++++ python/tests/server/voice_env.py | 9 +- python/timbal/server/README.md | 12 ++ python/timbal/server/http.py | 7 + python/timbal/server/recording_upload.py | 12 ++ python/timbal/server/rtc.py | 148 +++++++++++---- python/timbal/server/single_session.py | 150 ++++++++++++++++ python/timbal/server/voice.py | 11 ++ 10 files changed, 666 insertions(+), 32 deletions(-) create mode 100644 python/tests/server/test_rtc_force_relay.py create mode 100644 python/tests/server/test_single_session.py create mode 100644 python/timbal/server/single_session.py diff --git a/python/tests/server/test_rtc_force_relay.py b/python/tests/server/test_rtc_force_relay.py new file mode 100644 index 00000000..57a773c3 --- /dev/null +++ b/python/tests/server/test_rtc_force_relay.py @@ -0,0 +1,86 @@ +"""Tests for ``TIMBAL_VOICE_RTC_FORCE_RELAY`` — relay-only ICE for private-subnet boxes. + +Pure SDP/env logic; only the ``_ice_servers`` test needs aiortc. +""" + +from __future__ import annotations + +import pytest +from timbal.server.rtc import _force_relay, _ice_servers, _strip_non_relay_candidates + +from .voice_env import VOICE_ENV_KEYS + +_HOST = "a=candidate:6815297761 1 udp 2130706431 10.0.1.5 40000 typ host" +_SRFLX = "a=candidate:2932157868 1 udp 1694498815 34.1.2.3 40001 typ srflx raddr 10.0.1.5 rport 40000" +_RELAY = "a=candidate:3456 1 udp 25108223 52.4.5.6 50000 typ relay raddr 34.1.2.3 rport 40001" + + +def _sdp(*candidates: str) -> str: + return "\r\n".join( + [ + "v=0", + "o=- 1 1 IN IP4 0.0.0.0", + "s=-", + "t=0 0", + "m=audio 40000 UDP/TLS/RTP/SAVPF 96", + "c=IN IP4 10.0.1.5", + *candidates, + "a=sendrecv", + "", + ] + ) + + +@pytest.fixture(autouse=True) +def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None: + for k in VOICE_ENV_KEYS: + monkeypatch.delenv(k, raising=False) + monkeypatch.delenv("TIMBAL_STUN_URL", raising=False) + + +class TestStripNonRelayCandidates: + def test_keeps_only_relay_candidates(self) -> None: + out = _strip_non_relay_candidates(_sdp(_HOST, _SRFLX, _RELAY)) + assert _RELAY in out + assert "typ host" not in out + assert "typ srflx" not in out + # Everything else survives untouched, CRLF included. + assert out == _sdp(_RELAY) + + def test_degrades_to_original_when_no_relay_candidate(self) -> None: + """A failed TURN allocation must not produce an unconnectable answer.""" + sdp = _sdp(_HOST, _SRFLX) + assert _strip_non_relay_candidates(sdp) == sdp + + def test_noop_on_relay_only_sdp(self) -> None: + sdp = _sdp(_RELAY) + assert _strip_non_relay_candidates(sdp) == sdp + + +class TestForceRelayFlag: + def test_off_by_default(self) -> None: + assert not _force_relay() + + def test_requires_turn_to_be_configured(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TIMBAL_VOICE_RTC_FORCE_RELAY", "1") + assert not _force_relay() # degrades loudly instead of answering with nothing + + monkeypatch.setenv("TIMBAL_TURN_URL", "turn:turn.timbal.ai:3478") + assert _force_relay() + + +class TestIceServers: + def test_relay_only_returns_just_the_turn_server(self, monkeypatch: pytest.MonkeyPatch) -> None: + pytest.importorskip("aiortc", reason="timbal[voice] extra (aiortc) not installed") + monkeypatch.setenv("TIMBAL_TURN_URL", "turn:turn.timbal.ai:3478") + monkeypatch.setenv("TIMBAL_TURN_USERNAME", "user") + monkeypatch.setenv("TIMBAL_TURN_PASSWORD", "secret") + + servers = _ice_servers(relay_only=True) + assert len(servers) == 1 + assert servers[0].urls == "turn:turn.timbal.ai:3478" + assert servers[0].username == "user" + assert servers[0].credential == "secret" + + # Without relay_only the default STUN server is included too. + assert len(_ice_servers()) == 2 diff --git a/python/tests/server/test_single_session.py b/python/tests/server/test_single_session.py new file mode 100644 index 00000000..39c078b3 --- /dev/null +++ b/python/tests/server/test_single_session.py @@ -0,0 +1,188 @@ +"""Tests for ``TIMBAL_VOICE_SINGLE_SESSION`` — one voice session per process, then exit. + +``_exit_process`` is monkeypatched everywhere (the real thing is +``os._exit(0)``); a finished guard therefore stays around, which is exactly +what the refuse-after-served assertions need. + +The WebSocket transport exercises the full claim → connect → finish flow +without the aiortc dependency; the WebRTC side of the guard is covered in +``test_voice_rtc.py``. +""" + +from __future__ import annotations + +import asyncio +import time +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect +from timbal.server.http import create_app +from timbal.server.single_session import SingleSessionGuard, init_single_session_guard +from timbal.voice import TranscriptEvent + +from .test_voice_ws import _collect_ws_messages, _make_stt_class, _make_tts_class, _write_agent_module +from .voice_env import VOICE_ENV_KEYS + + +@pytest.fixture +def exits(monkeypatch: pytest.MonkeyPatch) -> list[int]: + """Record exit codes instead of killing the test process.""" + calls: list[int] = [] + monkeypatch.setattr("timbal.server.single_session._exit_process", calls.append) + return calls + + +class TestSingleSessionGuard: + async def test_claim_is_exclusive_until_released(self, exits: list[int]) -> None: + guard = SingleSessionGuard(idle_exit_secs=60.0) + assert guard.claim() + assert not guard.claim() + guard.release() + assert guard.claim() + assert exits == [] + + async def test_finish_exits_zero_and_refuses_forever(self, exits: list[int]) -> None: + guard = SingleSessionGuard(idle_exit_secs=60.0) + assert guard.claim() + await guard.finish() + assert exits == [0] + assert not guard.claim() + # A release after finish must not reopen the slot. + guard.release() + assert not guard.claim() + # Idempotent: a second finish doesn't exit twice. + await guard.finish() + assert exits == [0] + + async def test_idle_timer_exits_when_nothing_connects(self, exits: list[int]) -> None: + guard = SingleSessionGuard(idle_exit_secs=0.05) + guard.start() + await asyncio.sleep(0.2) + assert exits == [0] + + async def test_idle_timer_fires_on_claimed_but_never_connected(self, exits: list[int]) -> None: + """An offer whose ICE never completes must not keep the box alive.""" + guard = SingleSessionGuard(idle_exit_secs=0.05) + guard.start() + assert guard.claim() + await asyncio.sleep(0.2) + assert exits == [0] + + async def test_mark_connected_disarms_the_idle_timer(self, exits: list[int]) -> None: + guard = SingleSessionGuard(idle_exit_secs=0.05) + guard.start() + guard.mark_connected() + await asyncio.sleep(0.2) + assert exits == [] + + async def test_finish_waits_for_recording_uploads(self, exits: list[int]) -> None: + """The exit must not race the platform PUT — the process holds the data.""" + from timbal.server import recording_upload + + uploaded = asyncio.Event() + + async def _slow_upload() -> None: + await asyncio.sleep(0.05) + uploaded.set() + + task = asyncio.create_task(_slow_upload()) + recording_upload._upload_tasks.add(task) + task.add_done_callback(recording_upload._upload_tasks.discard) + + guard = SingleSessionGuard(idle_exit_secs=60.0) + assert guard.claim() + await guard.finish() + assert uploaded.is_set() + assert exits == [0] + + +class TestInitSingleSessionGuard: + async def test_disabled_without_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("TIMBAL_VOICE_SINGLE_SESSION", raising=False) + assert init_single_session_guard() is None + + async def test_enabled_with_default_idle_window(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + monkeypatch.delenv("TIMBAL_VOICE_IDLE_EXIT_SECS", raising=False) + guard = init_single_session_guard() + assert guard is not None + assert guard.idle_exit_secs == 60.0 + guard.shutdown() + + async def test_idle_window_from_env_with_bad_value_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + monkeypatch.setenv("TIMBAL_VOICE_IDLE_EXIT_SECS", "5") + guard = init_single_session_guard() + assert guard is not None and guard.idle_exit_secs == 5.0 + guard.shutdown() + + monkeypatch.setenv("TIMBAL_VOICE_IDLE_EXIT_SECS", "not-a-number") + guard = init_single_session_guard() + assert guard is not None and guard.idle_exit_secs == 60.0 + guard.shutdown() + + +def _setup_ws_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + spec = _write_agent_module(tmp_path, responses=["Hi there!"]) + monkeypatch.setenv("TIMBAL_RUNNABLE", spec) + for k in VOICE_ENV_KEYS: + monkeypatch.delenv(k, raising=False) + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + monkeypatch.setattr( + "timbal.voice.elevenlabs.ElevenLabsRealtimeSTT", + _make_stt_class([TranscriptEvent(type="committed", text="Hello")]), + ) + monkeypatch.setattr("timbal.voice.elevenlabs.ElevenLabsStreamTTS", _make_tts_class()) + + +class TestSingleSessionOverWebSocket: + def test_one_session_then_exit_then_refuse( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, exits: list[int] + ) -> None: + _setup_ws_env(monkeypatch, tmp_path) + app = create_app() + with TestClient(app) as client: + with client.websocket_connect("/voice/ws") as ws: + ws.send_json({}) + messages = _collect_ws_messages(ws) + assert messages[-1]["type"] == "session_ended" + + # The handler's finally runs after the socket closes; poll briefly. + deadline = time.monotonic() + 5.0 + while not exits and time.monotonic() < deadline: + time.sleep(0.02) + assert exits == [0] + + # One process = one session: a second connect is refused. + with client.websocket_connect("/voice/ws") as ws2: + with pytest.raises(WebSocketDisconnect) as excinfo: + ws2.receive_json() + assert excinfo.value.code == 1008 + + def test_idle_exit_when_nobody_connects( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, exits: list[int] + ) -> None: + _setup_ws_env(monkeypatch, tmp_path) + monkeypatch.setenv("TIMBAL_VOICE_IDLE_EXIT_SECS", "0.1") + app = create_app() + with TestClient(app): + deadline = time.monotonic() + 5.0 + while not exits and time.monotonic() < deadline: + time.sleep(0.02) + assert exits == [0] + + def test_no_single_session_env_means_no_lifetime_management( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, exits: list[int] + ) -> None: + _setup_ws_env(monkeypatch, tmp_path) + monkeypatch.delenv("TIMBAL_VOICE_SINGLE_SESSION") + app = create_app() + with TestClient(app) as client: + for _ in range(2): # sequential sessions stay allowed + with client.websocket_connect("/voice/ws") as ws: + ws.send_json({}) + messages = _collect_ws_messages(ws) + assert messages[-1]["type"] == "session_ended" + assert exits == [] diff --git a/python/tests/server/test_voice_rtc.py b/python/tests/server/test_voice_rtc.py index 47c69653..b4b5925f 100644 --- a/python/tests/server/test_voice_rtc.py +++ b/python/tests/server/test_voice_rtc.py @@ -209,6 +209,81 @@ async def close(self) -> None: assert types[-1] == "session_ended" +class TestVoiceRtcSingleSession: + async def test_one_call_then_exit_zero_then_409( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """TIMBAL_VOICE_SINGLE_SESSION: serve the call, exit 0, refuse a second offer.""" + _setup_env(monkeypatch, tmp_path, responses=["Hi there!"]) + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + exits: list[int] = [] + monkeypatch.setattr("timbal.server.single_session._exit_process", exits.append) + monkeypatch.setattr( + "timbal.voice.elevenlabs.ElevenLabsRealtimeSTT", + _make_delayed_stt_class( + [(0.2, TranscriptEvent(type="committed", text="Hello"))], + end_after=1.5, + ), + ) + monkeypatch.setattr( + "timbal.voice.elevenlabs.ElevenLabsStreamTTS", + _make_tts_class(chunk=b"\x01\x02" * 800, num_chunks=2), + ) + + app = create_app() + with TestClient(app) as client: + messages, _ = await _rtc_call(client) + assert messages[-1]["type"] == "session_ended" + + # The driver's finally (→ guard.finish) runs after the pc closes. + for _ in range(100): + if exits: + break + await asyncio.sleep(0.05) + assert exits == [0] + + # One process = one session: the 409 fires before any SDP work. + resp = await asyncio.to_thread( + client.post, "/voice/rtc", json={"sdp": "v=0", "type": "offer"} + ) + assert resp.status_code == 409 + + async def test_409_while_a_session_is_live( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + _setup_env(monkeypatch, tmp_path) + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + exits: list[int] = [] + monkeypatch.setattr("timbal.server.single_session._exit_process", exits.append) + + app = create_app() + with TestClient(app) as client: + assert app.state.single_session_guard.claim() # a session is live + resp = await asyncio.to_thread( + client.post, "/voice/rtc", json={"sdp": "v=0", "type": "offer"} + ) + assert resp.status_code == 409 + assert exits == [] + + async def test_rejected_offer_releases_the_slot( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """A 400 offer never became a session — the slot must reopen.""" + _setup_env(monkeypatch, tmp_path) + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + exits: list[int] = [] + monkeypatch.setattr("timbal.server.single_session._exit_process", exits.append) + + app = create_app() + with TestClient(app) as client: + resp = await asyncio.to_thread( + client.post, "/voice/rtc", json={"sdp": "not sdp at all", "type": "offer"} + ) + assert resp.status_code == 400 + assert app.state.single_session_guard.claim() + assert exits == [] + + class TestVoiceRtcSignalingErrors: def test_rejects_body_without_an_offer(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: _setup_env(monkeypatch, tmp_path) diff --git a/python/tests/server/voice_env.py b/python/tests/server/voice_env.py index a2dc32dc..d8914c6a 100644 --- a/python/tests/server/voice_env.py +++ b/python/tests/server/voice_env.py @@ -2,7 +2,8 @@ from __future__ import annotations -# Env vars that affect ``default_voice_config_from_env`` / ``merge_voice_config``. +# Env vars that affect voice server behavior: ``default_voice_config_from_env`` +# / ``merge_voice_config``, the single-session lifetime, and RTC ICE. VOICE_ENV_KEYS = ( "TIMBAL_STT_PROVIDER", "TIMBAL_STT_MODEL", @@ -21,4 +22,10 @@ "TIMBAL_VOICE_FILLER_MODEL", "TIMBAL_VOICE_FILLER_DELAY_SECS", "TIMBAL_VOICE_FILLER_REPEAT_SECS", + "TIMBAL_VOICE_SINGLE_SESSION", + "TIMBAL_VOICE_IDLE_EXIT_SECS", + "TIMBAL_VOICE_RTC_FORCE_RELAY", + "TIMBAL_TURN_URL", + "TIMBAL_TURN_USERNAME", + "TIMBAL_TURN_PASSWORD", ) diff --git a/python/timbal/server/README.md b/python/timbal/server/README.md index 6992793b..394efb45 100644 --- a/python/timbal/server/README.md +++ b/python/timbal/server/README.md @@ -189,6 +189,7 @@ POST /voice/rtc 200 → { "sdp": "", "type": "answer" } 400 → bad offer / no audio track / runnable is not an Agent +409 → single-session server already served (or is serving) its one session 501 → timbal[voice] extra (aiortc) not installed ``` @@ -206,6 +207,17 @@ The server answers with the TTS audio track on the same m-line. Client teardown | `TIMBAL_STUN_URL` | STUN server; defaults to `stun:stun.l.google.com:19302`. Set to empty to disable (loopback/LAN). | | `TIMBAL_TURN_URL` | Optional TURN server for clients behind symmetric NATs. | | `TIMBAL_TURN_USERNAME` / `TIMBAL_TURN_PASSWORD` | TURN credentials. | +| `TIMBAL_VOICE_RTC_FORCE_RELAY` | `1` → relay-only ICE: allocate on the TURN server above, skip STUN, and strip host/srflx candidates from the answer SDP. For servers on private subnets (serverless boxes) where every non-relay candidate is unreachable dead weight that slows the browser's ICE convergence. Requires `TIMBAL_TURN_URL` — without it (or if the TURN allocation yields no relay candidate) the server logs an error and answers with all candidates instead of an unconnectable SDP. | + +## Single-session lifetime (serverless voice boxes) + +`TIMBAL_VOICE_SINGLE_SESSION=1` makes the server process serve **exactly one voice session and then exit 0** — for deployments that spawn one process per call and reap the box on process exit (the platform cannot see a WebRTC call: media flows browser ↔ TURN ↔ box, so the process must own its lifetime). Applies to both transports; whichever session arrives first (WS or RTC) owns the process. + +- **Exit on session end.** When the one session ends (peer connection `failed`/`closed`, WebSocket closed, or the session ends server-side), the process finalizes — including waiting for the [platform recording push](#call-recording) to finish, since the process is the only thing holding that data — and exits 0. +- **Exit if nobody ever connects.** If no media connection is established within `TIMBAL_VOICE_IDLE_EXIT_SECS` (default `60`) of server start, exit 0. The window runs boot → *media established*, so an offer whose ICE never completes also exits after the window. +- **Refuse a second session.** While a session is live or after one has been served: `POST /voice/rtc` → **409**, `/voice/ws` → close **1008**. An offer rejected with 400 (bad SDP, no audio track) never became a session and does not consume the slot. + +All lifetime exits are code 0; env is read at server start (never at import time), so CRIU-restored processes with late-injected env behave identically. ## Ambient background audio diff --git a/python/timbal/server/http.py b/python/timbal/server/http.py index 07cf3077..c65ec9c7 100644 --- a/python/timbal/server/http.py +++ b/python/timbal/server/http.py @@ -40,6 +40,11 @@ async def lifespan( app.state.runnable = runnable app.state.job_store = JobStore() app.state.voice_config = merge_voice_config(runnable) + # Serverless voice boxes: serve one session, then exit (env read here at + # server start — post-CRIU-restore — never at import time). + from .single_session import init_single_session_guard + + app.state.single_session_guard = init_single_session_guard() # Voice warmup off the boot path: pre-import the voice stack (and pre-load # the local turn-detection models when server-configured) so the first # voice session doesn't pay those costs. No-op-ish for non-voice usage. @@ -54,6 +59,8 @@ async def lifespan( finally: if warmup_task is not None and not warmup_task.done(): warmup_task.cancel() + if app.state.single_session_guard is not None: + app.state.single_session_guard.shutdown() def create_app() -> FastAPI: diff --git a/python/timbal/server/recording_upload.py b/python/timbal/server/recording_upload.py index 8fec01e5..b6ba0fed 100644 --- a/python/timbal/server/recording_upload.py +++ b/python/timbal/server/recording_upload.py @@ -51,6 +51,18 @@ _upload_tasks: set[asyncio.Task[Any]] = set() +async def drain_upload_tasks() -> None: + """Block until every in-flight recording upload has finished. + + Single-session boxes call this before exiting: the process is the only + thing holding the recording, so the exit must wait for the platform PUT. + Bounded by the upload's own retry budget (~1h), which sits under the + platform's hard session cap. + """ + while _upload_tasks: + await asyncio.gather(*list(_upload_tasks), return_exceptions=True) + + def _recording_backoff(attempt: int) -> float: """1, 5, 25, 125, then 300s flat — the agreed recording retry curve.""" return min(1.0 * 5.0**attempt, 300.0) diff --git a/python/timbal/server/rtc.py b/python/timbal/server/rtc.py index 34f8f232..a343cf7d 100644 --- a/python/timbal/server/rtc.py +++ b/python/timbal/server/rtc.py @@ -44,14 +44,39 @@ _pcs: set[Any] = set() _drivers: set[asyncio.Task] = set() +_TRUTHY = frozenset({"1", "true", "yes", "on"}) + + +def _force_relay() -> bool: + """``TIMBAL_VOICE_RTC_FORCE_RELAY=1`` *and* a TURN server is configured. + + Serverless boxes sit in private subnets: host candidates are private IPs + and srflx can't receive unsolicited inbound, so every non-relay candidate + is dead weight that slows the browser's ICE convergence. Without a TURN + server relay-only would leave *no* reachable candidate, so a misconfigured + flag degrades (loudly) to normal gathering instead of a broken answer. + """ + if os.environ.get("TIMBAL_VOICE_RTC_FORCE_RELAY", "").strip().lower() not in _TRUTHY: + return False + if not os.environ.get("TIMBAL_TURN_URL"): + logger.error( + "voice_rtc_force_relay_without_turn", + hint="TIMBAL_VOICE_RTC_FORCE_RELAY=1 requires TIMBAL_TURN_URL/" + "TIMBAL_TURN_USERNAME/TIMBAL_TURN_PASSWORD; serving all candidates", + ) + return False + return True + -def _ice_servers() -> list[Any]: +def _ice_servers(*, relay_only: bool = False) -> list[Any]: from aiortc import RTCIceServer servers = [] - # Empty TIMBAL_STUN_URL disables STUN (loopback/tests); unset keeps the default. + # Empty TIMBAL_STUN_URL disables STUN (loopback/tests); unset keeps the + # default. Relay-only skips STUN entirely — srflx would be filtered from + # the answer anyway, so gathering it just burns signaling budget. stun_url = os.environ.get("TIMBAL_STUN_URL", "stun:stun.l.google.com:19302") - if stun_url: + if stun_url and not relay_only: servers.append(RTCIceServer(urls=stun_url)) turn_url = os.environ.get("TIMBAL_TURN_URL") if turn_url: @@ -65,6 +90,41 @@ def _ice_servers() -> list[Any]: return servers +def _strip_non_relay_candidates(sdp: str) -> str: + """Drop host/srflx ``a=candidate`` lines from an answer, keeping relay only. + + aiortc has no ``iceTransportPolicy``: it always gathers host candidates, + so relay-only has to be enforced on the answer SDP. The browser only + forms pairs with candidates it was told about, so filtering here is + enough — aiortc's internal host candidates are never checked. + + Degrades to the unfiltered SDP when filtering would leave no candidates + (e.g. the TURN allocation failed): a slow answer beats an unconnectable + one. The check is global rather than per m-section because all sections + share one gather (BUNDLE) and therefore identical candidate sets. + """ + lines = sdp.split("\r\n") + kept: list[str] = [] + removed = relayed = 0 + for line in lines: + if line.startswith("a=candidate:"): + if " typ relay" in line: + relayed += 1 + else: + removed += 1 + continue + kept.append(line) + if removed and not relayed: + logger.warning( + "voice_rtc_force_relay_no_relay_candidates", + hint="TURN allocation yielded no relay candidates; answering with all candidates", + ) + return sdp + if removed: + logger.debug("voice_rtc_relay_filtered", removed=removed, relay=relayed) + return "\r\n".join(kept) + + @router.post("/rtc") async def voice_rtc(request: Request) -> JSONResponse: try: @@ -93,6 +153,14 @@ async def voice_rtc(request: Request) -> JSONResponse: logger.error("voice_rtc_rejected", reason="runnable is not an Agent", type=type(runnable).__name__) return JSONResponse(status_code=400, content={"error": "Voice requires an Agent runnable."}) + guard = getattr(request.app.state, "single_session_guard", None) + if guard is not None and not guard.claim(): + logger.info("voice_rtc_rejected", reason="single-session server already served its session") + return JSONResponse( + status_code=409, + content={"error": "Single-session server: a voice session was already served."}, + ) + defaults = getattr(request.app.state, "voice_config", None) or VoiceConfig() sample_rate = int(merge_client_voice_overrides(defaults, config).sample_rate) @@ -103,7 +171,8 @@ async def voice_rtc(request: Request) -> JSONResponse: meta = {"playback_acks": "native", "transport": "webrtc", **meta} session.recording_meta = meta - pc = RTCPeerConnection(configuration=RTCConfiguration(iceServers=_ice_servers())) + force_relay = _force_relay() + pc = RTCPeerConnection(configuration=RTCConfiguration(iceServers=_ice_servers(relay_only=force_relay))) _pcs.add(pc) mic_track: Any = None @@ -145,7 +214,11 @@ def on_datachannel(ch: Any) -> None: @pc.on("connectionstatechange") async def on_connectionstatechange() -> None: logger.debug("voice_rtc_connection_state", state=pc.connectionState) - if pc.connectionState in ("failed", "closed"): + if pc.connectionState == "connected" and guard is not None: + # Media established: the single-session idle-exit window (boot → + # media connected) no longer applies. + guard.mark_connected() + if pc.connectionState in ("failed", "closed", "disconnected"): # Client is gone: end the session now rather than waiting for the # STT provider to time out on silence. Idempotent when the driver # already closed it. @@ -155,6 +228,8 @@ async def on_connectionstatechange() -> None: await pc.setRemoteDescription(RTCSessionDescription(sdp=sdp, type="offer")) except Exception as e: _pcs.discard(pc) + if guard is not None: + guard.release() logger.warning("voice_rtc_bad_offer", error=str(e)) return JSONResponse(status_code=400, content={"error": f"Invalid SDP offer: {e}"}) @@ -162,6 +237,8 @@ async def on_connectionstatechange() -> None: _pcs.discard(pc) with contextlib.suppress(Exception): await pc.close() + if guard is not None: + guard.release() return JSONResponse(status_code=400, content={"error": "Offer must contain an audio track."}) # After setRemoteDescription, so the TTS track reuses the offer's audio @@ -176,30 +253,38 @@ async def _drive() -> None: # connectionstatechange, so this only times out on a client that # connected but never offered a data channel. try: - await asyncio.wait_for(channel_ready.wait(), timeout=15.0) - except TimeoutError: - logger.warning("voice_rtc_no_datachannel") - with contextlib.suppress(Exception): - await pc.close() - _pcs.discard(pc) - return - mic_pcm = track_to_pcm(mic_track, sample_rate=sample_rate) - try: - async with aclosing(session.run(mic_pcm)) as events: - async for event in events: - if isinstance(event, AudioOutput): - downlink.write(event.data) - continue - for payload in event_to_payloads(event, session, meta): - _dc_send(payload) - except Exception as e: - logger.error("voice_rtc_session_error", error=str(e), exc_info=True) + try: + await asyncio.wait_for(channel_ready.wait(), timeout=15.0) + except TimeoutError: + logger.warning("voice_rtc_no_datachannel") + with contextlib.suppress(Exception): + await pc.close() + _pcs.discard(pc) + return + mic_pcm = track_to_pcm(mic_track, sample_rate=sample_rate) + try: + async with aclosing(session.run(mic_pcm)) as events: + async for event in events: + if isinstance(event, AudioOutput): + downlink.write(event.data) + continue + for payload in event_to_payloads(event, session, meta): + _dc_send(payload) + except Exception as e: + logger.error("voice_rtc_session_error", error=str(e), exc_info=True) + finally: + downlink.stop() + with contextlib.suppress(Exception): + await pc.close() + _pcs.discard(pc) + logger.info("voice_rtc_disconnected") finally: - downlink.stop() - with contextlib.suppress(Exception): - await pc.close() - _pcs.discard(pc) - logger.info("voice_rtc_disconnected") + # Single-session box: this offer was the one session, however it + # ended (call finished, ICE never completed, no data channel) — + # finalize (recording already pushed by session cleanup above) + # and exit 0 so the platform reaps the box. + if guard is not None: + await guard.finish() driver = asyncio.create_task(_drive()) _drivers.add(driver) @@ -208,7 +293,8 @@ async def _drive() -> None: answer = await pc.createAnswer() # Completes ICE gathering before returning — the answer is complete. await pc.setLocalDescription(answer) + answer_sdp = pc.localDescription.sdp + if force_relay: + answer_sdp = _strip_non_relay_candidates(answer_sdp) logger.info("voice_rtc_connected") - return JSONResponse( - content={"sdp": pc.localDescription.sdp, "type": pc.localDescription.type} - ) + return JSONResponse(content={"sdp": answer_sdp, "type": pc.localDescription.type}) diff --git a/python/timbal/server/single_session.py b/python/timbal/server/single_session.py new file mode 100644 index 00000000..14cc9951 --- /dev/null +++ b/python/timbal/server/single_session.py @@ -0,0 +1,150 @@ +"""Single-session lifetime for serverless voice boxes (``TIMBAL_VOICE_SINGLE_SESSION=1``). + +On serverless workforce deployments the platform spawns one server process +per voice call and reaps the box when the process exits — it cannot see the +call itself (WebRTC media flows browser ↔ TURN relay ↔ box, with no tunneled +socket), so the process must own its lifetime: + +* Serve **exactly one** voice session (WebRTC or WebSocket), then exit 0. + The exit waits for the recording upload (``PUT …/sessions/{session_id}``) + to finish first — this process is the only thing holding that data. +* Exit 0 if no media connection is established within + ``TIMBAL_VOICE_IDLE_EXIT_SECS`` (default 60) of server start. The window + is boot → *media established*, so a browser that POSTs an offer and never + completes ICE also exits after the window. +* Refuse a second session while one is live or after one has been served: + 409 on ``POST /voice/rtc``, close 1008 on ``/voice/ws``. + +All exits here are code 0 — the platform logs non-zero exits as crashes. + +Env is read at server start (the lifespan calls :func:`init_single_session_guard`), +never at import time: serverless boxes CRIU-restore from a framework-warm +snapshot with the real env arriving at restore time. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import os +import sys + +import structlog + +logger = structlog.get_logger("timbal.server.single_session") + +_TRUTHY = frozenset({"1", "true", "yes", "on"}) + +_DEFAULT_IDLE_EXIT_SECS = 60.0 + + +def _exit_process(code: int) -> None: + """Hard process exit. + + ``os._exit`` rather than ``sys.exit``: a ``SystemExit`` raised inside an + asyncio task is swallowed by the event loop / uvicorn and the process + would linger. Everything worth finalizing (recording upload, peer + connection teardown) has already happened by the time this runs; flush + stdio so the last log lines reach the platform's log tail. + """ + with contextlib.suppress(Exception): + sys.stdout.flush() + sys.stderr.flush() + os._exit(code) + + +class SingleSessionGuard: + """One voice session per process: claim → connect → finish → exit. + + Shared by the WebSocket and WebRTC transports (whichever claims first + owns the process). Not thread-safe — everything runs on the server's + single event loop, where the plain-attribute state machine is atomic. + """ + + def __init__(self, *, idle_exit_secs: float) -> None: + self.idle_exit_secs = idle_exit_secs + self._claimed = False + self._connected = False + self._finished = False + self._idle_task: asyncio.Task | None = None + + def start(self) -> None: + """Arm the idle-exit timer. Requires a running event loop (lifespan).""" + self._idle_task = asyncio.create_task(self._idle_exit(), name="voice-single-session-idle-exit") + + def shutdown(self) -> None: + """Disarm the idle-exit timer (lifespan teardown / media connected).""" + if self._idle_task is not None: + self._idle_task.cancel() + self._idle_task = None + + async def _idle_exit(self) -> None: + await asyncio.sleep(self.idle_exit_secs) + # A claimed-but-never-connected session exits too: the window runs + # boot → media established, so an offer whose ICE never completes + # doesn't keep the box (and its concurrency slot) alive. + logger.info( + "voice_single_session_idle_exit", + idle_exit_secs=self.idle_exit_secs, + session_claimed=self._claimed, + ) + _exit_process(0) + + def claim(self) -> bool: + """Reserve the one session slot; False while live or after it was served.""" + if self._claimed or self._finished: + return False + self._claimed = True + return True + + def release(self) -> None: + """Undo a claim whose session never started (e.g. a rejected offer).""" + if not self._finished: + self._claimed = False + + def mark_connected(self) -> None: + """Media is established — disarm the idle-exit timer.""" + if not self._connected: + self._connected = True + self.shutdown() + + async def finish(self) -> None: + """The one session ended: finalize and exit 0. + + Waits for in-flight recording uploads before exiting — bounded by + the upload's own retry budget (~1h), under the platform's hard + session cap. Idempotent. + """ + if self._finished: + return + self._finished = True + self._claimed = True + self.shutdown() + from .recording_upload import drain_upload_tasks + + try: + await drain_upload_tasks() + except Exception as e: + logger.error("voice_single_session_drain_failed", error=str(e)) + logger.info("voice_single_session_exit") + _exit_process(0) + + +def init_single_session_guard() -> SingleSessionGuard | None: + """Build and arm the guard when ``TIMBAL_VOICE_SINGLE_SESSION`` is set, else None. + + Called from the server lifespan — i.e. at server start, after any CRIU + restore has delivered the real env. + """ + if os.environ.get("TIMBAL_VOICE_SINGLE_SESSION", "").strip().lower() not in _TRUTHY: + return None + raw = os.environ.get("TIMBAL_VOICE_IDLE_EXIT_SECS", "") + try: + idle_exit_secs = float(raw) if raw.strip() else _DEFAULT_IDLE_EXIT_SECS + except ValueError: + logger.warning("voice_single_session_bad_idle_exit_secs", value=raw) + idle_exit_secs = _DEFAULT_IDLE_EXIT_SECS + guard = SingleSessionGuard(idle_exit_secs=idle_exit_secs) + guard.start() + logger.info("voice_single_session_enabled", idle_exit_secs=idle_exit_secs) + return guard diff --git a/python/timbal/server/voice.py b/python/timbal/server/voice.py index 57656305..2df79955 100644 --- a/python/timbal/server/voice.py +++ b/python/timbal/server/voice.py @@ -695,6 +695,15 @@ async def voice_ws(ws: WebSocket) -> None: await ws.close(code=1008, reason="Voice requires an Agent runnable") return + guard = getattr(ws.app.state, "single_session_guard", None) + if guard is not None: + if not guard.claim(): + logger.info("voice_ws_rejected", reason="single-session server already served its session") + await ws.close(code=1008, reason="Single-session server: a voice session was already served") + return + # On this transport the socket *is* the media connection. + guard.mark_connected() + audio_queue: asyncio.Queue[bytes] = asyncio.Queue() # Read the config hello. Protocol frames ("playback" acks, "audio", @@ -827,3 +836,5 @@ async def _handle(event: VoiceSessionEvent) -> None: except Exception as e: logger.debug("voice_session_close_suppressed", error=str(e)) logger.info("voice_ws_disconnected") + if guard is not None: + await guard.finish() From 22af2e97c92417d2686435e8a197e63f619c9f62 Mon Sep 17 00:00:00 2001 From: berges99 Date: Wed, 29 Jul 2026 22:09:58 +0200 Subject: [PATCH 2/5] fix(server): harden single-session exit paths for WS, idle drain, and RTC ICE Guarantee finish() after WS claim so handshake/session-build failures still exit; route idle exit through finish() so in-flight recording uploads drain; stop treating WebRTC disconnected as terminal (transient ICE blips). --- python/tests/server/test_single_session.py | 43 ++++++++++++++++++++++ python/timbal/server/rtc.py | 7 +++- python/timbal/server/single_session.py | 9 ++++- python/timbal/server/voice.py | 33 ++++++++++++----- 4 files changed, 79 insertions(+), 13 deletions(-) diff --git a/python/tests/server/test_single_session.py b/python/tests/server/test_single_session.py index 39c078b3..64d9266d 100644 --- a/python/tests/server/test_single_session.py +++ b/python/tests/server/test_single_session.py @@ -12,6 +12,7 @@ from __future__ import annotations import asyncio +import contextlib import time from pathlib import Path @@ -70,6 +71,26 @@ async def test_idle_timer_fires_on_claimed_but_never_connected(self, exits: list await asyncio.sleep(0.2) assert exits == [0] + async def test_idle_exit_waits_for_recording_uploads(self, exits: list[int]) -> None: + """The idle path must exit through finish(): never race an in-flight PUT.""" + from timbal.server import recording_upload + + uploaded = asyncio.Event() + + async def _slow_upload() -> None: + await asyncio.sleep(0.1) + uploaded.set() + + task = asyncio.create_task(_slow_upload()) + recording_upload._upload_tasks.add(task) + task.add_done_callback(recording_upload._upload_tasks.discard) + + guard = SingleSessionGuard(idle_exit_secs=0.05) + guard.start() + await asyncio.sleep(0.4) + assert uploaded.is_set() + assert exits == [0] + async def test_mark_connected_disarms_the_idle_timer(self, exits: list[int]) -> None: guard = SingleSessionGuard(idle_exit_secs=0.05) guard.start() @@ -161,6 +182,28 @@ def test_one_session_then_exit_then_refuse( ws2.receive_json() assert excinfo.value.code == 1008 + def test_session_build_failure_still_exits( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, exits: list[int] + ) -> None: + """An exception between claim() and the session's own teardown must not + leave the process alive: the idle timer is already disarmed by then.""" + _setup_ws_env(monkeypatch, tmp_path) + + def _boom(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("recorder misconfigured") + + monkeypatch.setattr("timbal.server.voice.build_voice_session", _boom) + + app = create_app() + with TestClient(app, raise_server_exceptions=False) as client: + with contextlib.suppress(Exception), client.websocket_connect("/voice/ws") as ws: + ws.send_json({}) + ws.receive_json() + deadline = time.monotonic() + 5.0 + while not exits and time.monotonic() < deadline: + time.sleep(0.02) + assert exits == [0] + def test_idle_exit_when_nobody_connects( self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, exits: list[int] ) -> None: diff --git a/python/timbal/server/rtc.py b/python/timbal/server/rtc.py index a343cf7d..59a298fd 100644 --- a/python/timbal/server/rtc.py +++ b/python/timbal/server/rtc.py @@ -218,10 +218,13 @@ async def on_connectionstatechange() -> None: # Media established: the single-session idle-exit window (boot → # media connected) no longer applies. guard.mark_connected() - if pc.connectionState in ("failed", "closed", "disconnected"): + if pc.connectionState in ("failed", "closed"): # Client is gone: end the session now rather than waiting for the # STT provider to time out on silence. Idempotent when the driver - # already closed it. + # already closed it. Deliberately *not* "disconnected": per W3C + # semantics that's a possibly-transient ICE blip that can return + # to "connected" (aiortc doesn't emit it today, but if it ever + # does, tearing down a recoverable call would be wrong). await session.close() try: diff --git a/python/timbal/server/single_session.py b/python/timbal/server/single_session.py index 14cc9951..3ba96d2d 100644 --- a/python/timbal/server/single_session.py +++ b/python/timbal/server/single_session.py @@ -88,7 +88,14 @@ async def _idle_exit(self) -> None: idle_exit_secs=self.idle_exit_secs, session_claimed=self._claimed, ) - _exit_process(0) + # Detach before finish(): its shutdown() cancels self._idle_task, + # which is this very task — cancelling ourselves would abort the + # upload drain mid-flight. + self._idle_task = None + # Exit through finish(), not a bare _exit_process: the exit must + # never race an in-flight recording upload, even in states that + # shouldn't leave one behind while the timer is still armed. + await self.finish() def claim(self) -> bool: """Reserve the one session slot; False while live or after it was served.""" diff --git a/python/timbal/server/voice.py b/python/timbal/server/voice.py index 2df79955..c59be04e 100644 --- a/python/timbal/server/voice.py +++ b/python/timbal/server/voice.py @@ -684,7 +684,6 @@ def event_to_payloads(event: Any, session: Any, meta: dict[str, Any]) -> list[di @router.websocket("/ws") async def voice_ws(ws: WebSocket) -> None: from ..core.agent import Agent - from ..voice import SessionInterrupted, VoiceSessionEvent await ws.accept() logger.info("voice_ws_connected") @@ -696,13 +695,29 @@ async def voice_ws(ws: WebSocket) -> None: return guard = getattr(ws.app.state, "single_session_guard", None) - if guard is not None: - if not guard.claim(): - logger.info("voice_ws_rejected", reason="single-session server already served its session") - await ws.close(code=1008, reason="Single-session server: a voice session was already served") - return - # On this transport the socket *is* the media connection. - guard.mark_connected() + if guard is None: + await _serve_voice_ws(ws, runnable) + return + + if not guard.claim(): + logger.info("voice_ws_rejected", reason="single-session server already served its session") + await ws.close(code=1008, reason="Single-session server: a voice session was already served") + return + # On this transport the socket *is* the media connection. + guard.mark_connected() + try: + await _serve_voice_ws(ws, runnable) + finally: + # However the session ended — including an exception in the handshake + # or session build — this socket was the one session. Without this, + # a failure between claim() (idle timer disarmed) and the session's + # own teardown would leave the process alive forever. + await guard.finish() + + +async def _serve_voice_ws(ws: WebSocket, runnable: Any) -> None: + """One WebSocket voice session: hello → session → events, until close.""" + from ..voice import SessionInterrupted, VoiceSessionEvent audio_queue: asyncio.Queue[bytes] = asyncio.Queue() @@ -836,5 +851,3 @@ async def _handle(event: VoiceSessionEvent) -> None: except Exception as e: logger.debug("voice_session_close_suppressed", error=str(e)) logger.info("voice_ws_disconnected") - if guard is not None: - await guard.finish() From 9a14b4694916a2e0605d4125ada446bd8e2420bf Mon Sep 17 00:00:00 2001 From: berges99 Date: Wed, 29 Jul 2026 22:49:53 +0200 Subject: [PATCH 3/5] fix(server): release single-session slot on RTC setup crashes A claim() that fails before the driver task starts (e.g. build_voice_session) left the slot claimed with no finish()/release, so retries got 409 until idle exit. Mirror the 400 paths: release on setup exceptions and re-raise the 500. --- python/tests/server/test_voice_rtc.py | 24 ++++ python/timbal/server/rtc.py | 181 ++++++++++++++------------ 2 files changed, 123 insertions(+), 82 deletions(-) diff --git a/python/tests/server/test_voice_rtc.py b/python/tests/server/test_voice_rtc.py index b4b5925f..2c342e35 100644 --- a/python/tests/server/test_voice_rtc.py +++ b/python/tests/server/test_voice_rtc.py @@ -265,6 +265,30 @@ async def test_409_while_a_session_is_live( assert resp.status_code == 409 assert exits == [] + async def test_setup_crash_releases_the_slot( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path + ) -> None: + """A 500 between claim() and the driver task must not park the box in a + claimed state: the slot reopens and the idle timer bounds the lifetime.""" + _setup_env(monkeypatch, tmp_path) + monkeypatch.setenv("TIMBAL_VOICE_SINGLE_SESSION", "1") + exits: list[int] = [] + monkeypatch.setattr("timbal.server.single_session._exit_process", exits.append) + + def _boom(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("recorder misconfigured") + + monkeypatch.setattr("timbal.server.rtc.build_voice_session", _boom) + + app = create_app() + with TestClient(app, raise_server_exceptions=False) as client: + resp = await asyncio.to_thread( + client.post, "/voice/rtc", json={"sdp": "v=0", "type": "offer"} + ) + assert resp.status_code == 500 + assert app.state.single_session_guard.claim() + assert exits == [] + async def test_rejected_offer_releases_the_slot( self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: diff --git a/python/timbal/server/rtc.py b/python/timbal/server/rtc.py index 59a298fd..cd07d68e 100644 --- a/python/timbal/server/rtc.py +++ b/python/timbal/server/rtc.py @@ -161,92 +161,109 @@ async def voice_rtc(request: Request) -> JSONResponse: content={"error": "Single-session server: a voice session was already served."}, ) - defaults = getattr(request.app.state, "voice_config", None) or VoiceConfig() - sample_rate = int(merge_client_voice_overrides(defaults, config).sample_rate) - - downlink = PcmQueueTrack(sample_rate=sample_rate) - session, meta = build_voice_session( - runnable, defaults, config, playback_tracker=PacedPlaybackTracker(downlink) - ) - meta = {"playback_acks": "native", "transport": "webrtc", **meta} - session.recording_meta = meta - - force_relay = _force_relay() - pc = RTCPeerConnection(configuration=RTCConfiguration(iceServers=_ice_servers(relay_only=force_relay))) - _pcs.add(pc) - - mic_track: Any = None - channel: Any = None - channel_ready = asyncio.Event() - # Session events can fire before SCTP is up; buffer until the client's - # data channel arrives. - pending_payloads: list[str] = [] - - def _dc_send(payload: dict) -> None: - msg = json.dumps(payload) - if channel is not None and channel.readyState == "open": - try: - channel.send(msg) - except Exception as e: - logger.debug("voice_rtc_dc_send_failed", error=str(e), msg_type=payload.get("type")) - else: - pending_payloads.append(msg) - - @pc.on("track") - def on_track(track: Any) -> None: - nonlocal mic_track - if track.kind == "audio" and mic_track is None: - mic_track = track - - @pc.on("datachannel") - def on_datachannel(ch: Any) -> None: - nonlocal channel - channel = ch - for msg in pending_payloads: - try: - ch.send(msg) - except Exception as e: - logger.debug("voice_rtc_dc_flush_failed", error=str(e)) - break - pending_payloads.clear() - channel_ready.set() - - @pc.on("connectionstatechange") - async def on_connectionstatechange() -> None: - logger.debug("voice_rtc_connection_state", state=pc.connectionState) - if pc.connectionState == "connected" and guard is not None: - # Media established: the single-session idle-exit window (boot → - # media connected) no longer applies. - guard.mark_connected() - if pc.connectionState in ("failed", "closed"): - # Client is gone: end the session now rather than waiting for the - # STT provider to time out on silence. Idempotent when the driver - # already closed it. Deliberately *not* "disconnected": per W3C - # semantics that's a possibly-transient ICE blip that can return - # to "connected" (aiortc doesn't emit it today, but if it ever - # does, tearing down a recoverable call would be wrong). - await session.close() - + pc: Any = None try: - await pc.setRemoteDescription(RTCSessionDescription(sdp=sdp, type="offer")) - except Exception as e: - _pcs.discard(pc) - if guard is not None: - guard.release() - logger.warning("voice_rtc_bad_offer", error=str(e)) - return JSONResponse(status_code=400, content={"error": f"Invalid SDP offer: {e}"}) + defaults = getattr(request.app.state, "voice_config", None) or VoiceConfig() + sample_rate = int(merge_client_voice_overrides(defaults, config).sample_rate) - if mic_track is None: - _pcs.discard(pc) - with contextlib.suppress(Exception): - await pc.close() + downlink = PcmQueueTrack(sample_rate=sample_rate) + session, meta = build_voice_session( + runnable, defaults, config, playback_tracker=PacedPlaybackTracker(downlink) + ) + meta = {"playback_acks": "native", "transport": "webrtc", **meta} + session.recording_meta = meta + + force_relay = _force_relay() + pc = RTCPeerConnection(configuration=RTCConfiguration(iceServers=_ice_servers(relay_only=force_relay))) + _pcs.add(pc) + + mic_track: Any = None + channel: Any = None + channel_ready = asyncio.Event() + # Session events can fire before SCTP is up; buffer until the client's + # data channel arrives. + pending_payloads: list[str] = [] + + def _dc_send(payload: dict) -> None: + msg = json.dumps(payload) + if channel is not None and channel.readyState == "open": + try: + channel.send(msg) + except Exception as e: + logger.debug("voice_rtc_dc_send_failed", error=str(e), msg_type=payload.get("type")) + else: + pending_payloads.append(msg) + + @pc.on("track") + def on_track(track: Any) -> None: + nonlocal mic_track + if track.kind == "audio" and mic_track is None: + mic_track = track + + @pc.on("datachannel") + def on_datachannel(ch: Any) -> None: + nonlocal channel + channel = ch + for msg in pending_payloads: + try: + ch.send(msg) + except Exception as e: + logger.debug("voice_rtc_dc_flush_failed", error=str(e)) + break + pending_payloads.clear() + channel_ready.set() + + @pc.on("connectionstatechange") + async def on_connectionstatechange() -> None: + logger.debug("voice_rtc_connection_state", state=pc.connectionState) + if pc.connectionState == "connected" and guard is not None: + # Media established: the single-session idle-exit window (boot → + # media connected) no longer applies. + guard.mark_connected() + if pc.connectionState in ("failed", "closed"): + # Client is gone: end the session now rather than waiting for the + # STT provider to time out on silence. Idempotent when the driver + # already closed it. Deliberately *not* "disconnected": per W3C + # semantics that's a possibly-transient ICE blip that can return + # to "connected" (aiortc doesn't emit it today, but if it ever + # does, tearing down a recoverable call would be wrong). + await session.close() + + try: + await pc.setRemoteDescription(RTCSessionDescription(sdp=sdp, type="offer")) + except Exception as e: + _pcs.discard(pc) + if guard is not None: + guard.release() + logger.warning("voice_rtc_bad_offer", error=str(e)) + return JSONResponse(status_code=400, content={"error": f"Invalid SDP offer: {e}"}) + + if mic_track is None: + _pcs.discard(pc) + with contextlib.suppress(Exception): + await pc.close() + if guard is not None: + guard.release() + return JSONResponse(status_code=400, content={"error": "Offer must contain an audio track."}) + + # After setRemoteDescription, so the TTS track reuses the offer's audio + # transceiver instead of adding a second m-line the client never offered. + pc.addTrack(downlink) + except Exception: + # A setup crash after claim() (build_voice_session, pc construction, + # addTrack) must not park the box in a claimed state with no + # finish(): reopen the slot and let the 500 out — the idle timer is + # still armed (media never connected), so the box lifetime stays + # bounded. Once the driver task below exists, lifetime is its + # finish()'s responsibility instead, never a release. The 400 paths + # above return (not raise) and do their own release. + if pc is not None: + _pcs.discard(pc) + with contextlib.suppress(Exception): + await pc.close() if guard is not None: guard.release() - return JSONResponse(status_code=400, content={"error": "Offer must contain an audio track."}) - - # After setRemoteDescription, so the TTS track reuses the offer's audio - # transceiver instead of adding a second m-line the client never offered. - pc.addTrack(downlink) + raise async def _drive() -> None: # Do not start the session before the transport is usable: STT starts From a346054dda0f42693c0ebdbabd5abc289f908415 Mon Sep 17 00:00:00 2001 From: berges99 Date: Wed, 29 Jul 2026 22:54:20 +0200 Subject: [PATCH 4/5] fix(server): don't idle-exit a single-session box after media connects _idle_exit cleared the timer task and always finish()'d without checking _connected, so a WebRTC connect at the idle deadline could still exit 0. Yield once for queued ICE handlers, re-check through the upload drain, and only then commit the exit. --- python/tests/server/test_single_session.py | 41 ++++++++++++++++++++++ python/timbal/server/single_session.py | 34 ++++++++++++++---- 2 files changed, 68 insertions(+), 7 deletions(-) diff --git a/python/tests/server/test_single_session.py b/python/tests/server/test_single_session.py index 64d9266d..84347057 100644 --- a/python/tests/server/test_single_session.py +++ b/python/tests/server/test_single_session.py @@ -98,6 +98,47 @@ async def test_mark_connected_disarms_the_idle_timer(self, exits: list[int]) -> await asyncio.sleep(0.2) assert exits == [] + async def test_idle_exit_aborts_if_connected_at_deadline(self, exits: list[int]) -> None: + """Media establishing in the same turn as the timer must not kill the call. + + After sleep wakes, _idle_exit yields once so a queued mark_connected() + can run before exit; clearing _idle_task must not strand a late + connect with an inevitable exit. + """ + guard = SingleSessionGuard(idle_exit_secs=0.05) + guard.start() + assert guard.claim() + + async def _connect_at_deadline() -> None: + await asyncio.sleep(0.05) + guard.mark_connected() + + asyncio.create_task(_connect_at_deadline()) + await asyncio.sleep(0.3) + assert exits == [] + assert guard._connected + assert not guard._finished + + async def test_idle_exit_aborts_if_connected_during_drain( + self, exits: list[int], monkeypatch: pytest.MonkeyPatch + ) -> None: + """mark_connected during the idle upload drain must not exit the box.""" + guard = SingleSessionGuard(idle_exit_secs=0.05) + + async def _drain_and_connect() -> None: + guard.mark_connected() + + monkeypatch.setattr( + "timbal.server.recording_upload.drain_upload_tasks", + _drain_and_connect, + ) + guard.start() + assert guard.claim() + await asyncio.sleep(0.3) + assert exits == [] + assert guard._connected + assert not guard._finished + async def test_finish_waits_for_recording_uploads(self, exits: list[int]) -> None: """The exit must not race the platform PUT — the process holds the data.""" from timbal.server import recording_upload diff --git a/python/timbal/server/single_session.py b/python/timbal/server/single_session.py index 3ba96d2d..2f92958d 100644 --- a/python/timbal/server/single_session.py +++ b/python/timbal/server/single_session.py @@ -80,6 +80,14 @@ def shutdown(self) -> None: async def _idle_exit(self) -> None: await asyncio.sleep(self.idle_exit_secs) + # ICE can complete in the same event-loop turn as the timer firing. + # Yield once so a queued connectionstatechange → mark_connected() + # runs before we decide the box is unused; without this, clearing + # _idle_task below would make a late mark_connected() a no-op and + # finish() would kill a call that should count as media established. + await asyncio.sleep(0) + if self._connected or self._finished: + return # A claimed-but-never-connected session exits too: the window runs # boot → media established, so an offer whose ICE never completes # doesn't keep the box (and its concurrency slot) alive. @@ -88,14 +96,26 @@ async def _idle_exit(self) -> None: idle_exit_secs=self.idle_exit_secs, session_claimed=self._claimed, ) - # Detach before finish(): its shutdown() cancels self._idle_task, - # which is this very task — cancelling ourselves would abort the - # upload drain mid-flight. + # Detach before awaiting so shutdown()/finish() cannot cancel this + # very task mid-drain. Commit _finished only after the final + # _connected check — media may still land while we drain uploads. self._idle_task = None - # Exit through finish(), not a bare _exit_process: the exit must - # never race an in-flight recording upload, even in states that - # shouldn't leave one behind while the timer is still armed. - await self.finish() + if self._connected or self._finished: + return + from .recording_upload import drain_upload_tasks + + try: + await drain_upload_tasks() + except Exception as e: + logger.error("voice_single_session_drain_failed", error=str(e)) + if self._connected or self._finished: + # Connected (or the session already finished) while draining — + # session teardown owns the process lifetime from here. + return + self._finished = True + self._claimed = True + logger.info("voice_single_session_exit") + _exit_process(0) def claim(self) -> bool: """Reserve the one session slot; False while live or after it was served.""" From 333f4532db9916fe12f951da4197dbdb7676efbb Mon Sep 17 00:00:00 2001 From: berges99 Date: Wed, 29 Jul 2026 23:40:36 +0200 Subject: [PATCH 5/5] fix(voice): use threading.Lock for shared Smart Turn / Namo model load asyncio.Lock on the process-wide EOU singletons breaks under Starlette TestClient on Windows/py3.13: lifespan warmup binds the lock (often while stuck on HF 429), then the WS handler runs on another loop. Load is already executor work, so a thread lock is the right primitive. --- python/timbal/voice/namo.py | 16 ++++++++++++---- python/timbal/voice/smart_turn.py | 21 +++++++++++++++++---- 2 files changed, 29 insertions(+), 8 deletions(-) diff --git a/python/timbal/voice/namo.py b/python/timbal/voice/namo.py index 1d2f3720..6b8d3a69 100644 --- a/python/timbal/voice/namo.py +++ b/python/timbal/voice/namo.py @@ -39,6 +39,7 @@ import asyncio import os import re +import threading from functools import lru_cache, partial import numpy as np @@ -183,14 +184,21 @@ def __init__( self.max_length = _MAX_LENGTH_BY_REPO.get(self.repo_id, _DEFAULT_MAX_LENGTH) self._session: ort.InferenceSession | None = None self._tokenizer: object | None = None - self._load_lock = asyncio.Lock() + # threading.Lock — same rationale as SmartTurnEouModel: process-wide + # singleton reused across TestClient lifespan vs request event loops. + self._load_lock = threading.Lock() async def start(self) -> None: - async with self._load_lock: + if self._session is not None: + return + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, self._ensure_loaded) + + def _ensure_loaded(self) -> None: + with self._load_lock: if self._session is not None: return - loop = asyncio.get_running_loop() - self._session, self._tokenizer = await loop.run_in_executor(None, self._load) + self._session, self._tokenizer = self._load() async def close(self) -> None: # Shared process-wide; keep alive. diff --git a/python/timbal/voice/smart_turn.py b/python/timbal/voice/smart_turn.py index 10d28e4a..36295d65 100644 --- a/python/timbal/voice/smart_turn.py +++ b/python/timbal/voice/smart_turn.py @@ -39,6 +39,7 @@ import asyncio import os +import threading from functools import lru_cache, partial import numpy as np @@ -161,15 +162,27 @@ def __init__( self.cpu_count = cpu_count self._session: ort.InferenceSession | None = None self._sample_rate = _MODEL_SAMPLE_RATE - self._load_lock = asyncio.Lock() + # threading.Lock — not asyncio.Lock: this model is a process-wide + # singleton shared across sessions, and Starlette's TestClient runs + # lifespan warmup on a different event loop than request handlers + # (Windows + py3.13 surfaces it). An asyncio.Lock bound during a + # HF-download warmup then crashes the first WS session with + # "bound to a different event loop". The load itself is sync work + # offloaded to the executor, so a thread lock is the right primitive. + self._load_lock = threading.Lock() async def start(self, *, sample_rate: int) -> None: self._sample_rate = int(sample_rate) or _MODEL_SAMPLE_RATE - async with self._load_lock: + if self._session is not None: + return + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, self._ensure_loaded) + + def _ensure_loaded(self) -> None: + with self._load_lock: if self._session is not None: return - loop = asyncio.get_running_loop() - self._session = await loop.run_in_executor(None, self._load_session) + self._session = self._load_session() async def close(self) -> None: # Shared across sessions; the ort session holds no per-run state and