feat(server): single-session lifetime and relay-only ICE for serverle…

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

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 == []

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)

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",
 )