Feat/voice robustness

examples/voice_turn_modes.py

@@ -18,21 +18,41 @@
 
 from __future__ import annotations
 
+import asyncio
 import os
+import random
+from datetime import datetime, timezone
 
 from timbal import Agent
+from timbal.core.tool import Tool
 from timbal.voice import resolve_turn_detector
 
 _MODE = os.environ.get("TIMBAL_VOICE_TURN_DETECTOR", "heuristic").strip().lower()
 
+
+async def get_datetime() -> str:
+    """Get the current UTC date and time (slow on purpose for voice tool-call testing)."""
+    await asyncio.sleep(random.uniform(3.0, 5.0))
+    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S %Z")
+
+
 agent = Agent(
     name="voice_turn_modes",
-    model=os.environ.get("TIMBAL_VOICE_DEMO_MODEL", "openai/gpt-4o-mini"),
+    model=os.environ.get("TIMBAL_VOICE_DEMO_MODEL", "groq/llama-3.1-8b-instant"),
+    max_tokens=1024,  # required when switching to Anthropic from the playground
     system_prompt=(
         "You are a concise voice assistant. Keep replies to 1–2 short sentences. "
+        "When the user asks for the date and/or time, you MUST call get_datetime "
+        "and then speak the returned value clearly (include both date and time). "
+        "Never claim you cannot tell the time. "
         f"(turn_detector mode: {_MODE})"
     ),
-    tools=[],
+    tools=[
+        Tool(
+            handler=get_datetime,
+            description="Return the current UTC date and time. Always use this for date/time questions.",
+        )
+    ],
 )
 
 # Server reads this on startup (instance or mode name). Client WS JSON cannot override it.

pyproject.toml

@@ -63,6 +63,7 @@ server = [
 voice = [
   "huggingface-hub>=0.28.0",
   "onnxruntime>=1.20.0",
+  "transformers>=4.40.0",
 ]
 all = [
   "timbal[codegen]",

python/tests/server/test_voice_config.py

@@ -24,6 +24,7 @@
 class TestDefaultVoiceConfigFromEnv:
     def test_defaults_when_unset(self) -> None:
         cfg = voice_routes.default_voice_config_from_env()
+        assert cfg["stt_provider"] == "elevenlabs"
         assert cfg["stt_model"] == "scribe_v2_realtime"
         assert cfg["tts_model"] == "eleven_flash_v2_5"
         assert cfg["voice"] == voice_routes._DEFAULT_VOICE_ID
@@ -33,10 +34,12 @@ def test_defaults_when_unset(self) -> None:
         assert cfg["tts_extra"]["auto_mode"] is True
 
     def test_env_overrides(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TIMBAL_STT_PROVIDER", "deepgram")
         monkeypatch.setenv("TIMBAL_STT_MODEL", "custom_stt")
         monkeypatch.setenv("TIMBAL_TTS_MODEL", "custom_tts")
         monkeypatch.setenv("TIMBAL_VOICE_LANGUAGE", "en")
         cfg = voice_routes.default_voice_config_from_env()
+        assert cfg["stt_provider"] == "deepgram"
         assert cfg["stt_model"] == "custom_stt"
         assert cfg["tts_model"] == "custom_tts"
         assert cfg["language"] == "en"

python/tests/server/voice_env.py

@@ -4,6 +4,7 @@
 
 # Env vars that affect ``default_voice_config_from_env`` / ``merge_voice_config``.
 VOICE_ENV_KEYS = (
+    "TIMBAL_STT_PROVIDER",
     "TIMBAL_STT_MODEL",
     "TIMBAL_TTS_MODEL",
     "ELEVENLABS_VOICE_ID",

python/tests/utils/test_serialization.py

@@ -169,6 +169,16 @@ def test_empty_string_returns_empty_dict(self):
         assert coerce_to_dict("") == {}
         assert coerce_to_dict("   ") == {}
 
+    def test_nullish_no_arg_tool_inputs(self):
+        # Groq / OpenAI-compat often emit null for tools with zero parameters.
+        assert coerce_to_dict(None) == {}
+        assert coerce_to_dict("null") == {}
+        assert coerce_to_dict("None") == {}
+        assert coerce_to_dict("NULL") == {}
+
+    def test_json_null_literal_returns_empty_dict(self):
+        assert coerce_to_dict("null") == {}
+
     def test_literal_eval_fallback(self):
         # Python dict literal that isn't valid JSON (single quotes)
         assert coerce_to_dict("{'a': 1}") == {"a": 1}
@@ -177,6 +187,10 @@ def test_unparseable_string_raises(self):
         with pytest.raises(ValueError, match="Cannot coerce value to dict"):
             coerce_to_dict("not a dict at all!")
 
+    def test_non_dict_json_raises(self):
+        with pytest.raises(ValueError, match="Cannot coerce value to dict"):
+            coerce_to_dict("[1, 2, 3]")
+
     def test_non_string_non_dict_raises(self):
         with pytest.raises(ValueError, match="Cannot coerce value to dict"):
             coerce_to_dict(42)

python/tests/voice/test_deepgram.py

@@ -0,0 +1,278 @@
+"""Deepgram STT providers: message → TranscriptEvent mapping, URI building, resolve_stt.
+
+No live API calls — messages are fed straight into ``_handle_message`` and the
+event queue is drained, mirroring how the receive loop feeds ``events()``.
+"""
+
+from __future__ import annotations
+
+import json
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+from timbal.voice.deepgram import (
+    DEFAULT_FLUX_MODEL,
+    DEFAULT_NOVA_MODEL,
+    DeepgramFluxSTT,
+    DeepgramNovaSTT,
+    effective_stt_model,
+    is_flux_model,
+    resolve_stt,
+)
+from timbal.voice.elevenlabs import ElevenLabsRealtimeSTT
+from timbal.voice.session import AudioInputConfig
+
+
+def _drain(stt) -> list:
+    events = []
+    while not stt._queue.empty():
+        events.append(stt._queue.get_nowait())
+    return events
+
+
+def _flux_turn_info(event: str, transcript: str = "hello there", **kw) -> dict:
+    return {
+        "type": "TurnInfo",
+        "event": event,
+        "transcript": transcript,
+        "turn_index": 0,
+        "end_of_turn_confidence": 0.9,
+        **kw,
+    }
+
+
+class _FakeWs:
+    def __init__(self) -> None:
+        self.sent: list = []
+
+    async def send(self, payload) -> None:
+        self.sent.append(payload)
+
+
+class TestFluxEventMapping:
+    async def test_update_and_start_of_turn_are_partials(self) -> None:
+        stt = DeepgramFluxSTT()
+        await stt._handle_message(_flux_turn_info("StartOfTurn", "hi"))
+        await stt._handle_message(_flux_turn_info("Update", "hi there"))
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [("partial", "hi"), ("partial", "hi there")]
+
+    async def test_end_of_turn_is_committed(self) -> None:
+        stt = DeepgramFluxSTT()
+        await stt._handle_message(_flux_turn_info("EndOfTurn", "tell me a story."))
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [("committed", "tell me a story.")]
+
+    async def test_eager_events_emit_nothing(self) -> None:
+        stt = DeepgramFluxSTT()
+        await stt._handle_message(_flux_turn_info("EagerEndOfTurn"))
+        await stt._handle_message(_flux_turn_info("TurnResumed"))
+        assert _drain(stt) == []
+
+    async def test_empty_transcripts_skipped(self) -> None:
+        stt = DeepgramFluxSTT()
+        await stt._handle_message(_flux_turn_info("Update", ""))
+        await stt._handle_message(_flux_turn_info("EndOfTurn", "  "))
+        assert _drain(stt) == []
+
+    async def test_connected_ignored_fatal_error_surfaces(self) -> None:
+        stt = DeepgramFluxSTT()
+        await stt._handle_message({"type": "Connected", "request_id": "r1", "sequence_id": 0})
+        await stt._handle_message({"type": "FatalError", "description": "bad auth", "code": "AUTH"})
+        events = _drain(stt)
+        assert len(events) == 1
+        assert events[0].type == "error"
+        assert "bad auth" in events[0].text
+
+    async def test_commit_is_noop(self) -> None:
+        stt = DeepgramFluxSTT()
+        stt._ws = _FakeWs()
+        await stt.commit()
+        assert stt._ws.sent == []
+
+
+class TestFluxUri:
+    def _parse(self, stt: DeepgramFluxSTT, config: AudioInputConfig) -> dict:
+        uri = stt._build_uri(config)
+        parsed = urlparse(uri)
+        assert parsed.scheme == "wss"
+        assert parsed.path == "/v2/listen"
+        return parse_qs(parsed.query)
+
+    def test_defaults(self) -> None:
+        q = self._parse(DeepgramFluxSTT(), AudioInputConfig())
+        assert q["model"] == [DEFAULT_FLUX_MODEL]
+        assert q["encoding"] == ["linear16"]
+        assert q["sample_rate"] == ["16000"]
+        assert q["eot_timeout_ms"] == ["2000"]
+        assert "language" not in q
+
+    def test_language_hint_only_for_multi(self) -> None:
+        q = self._parse(DeepgramFluxSTT(), AudioInputConfig(language="es"))
+        assert q["language_hint"] == ["es"]
+        q = self._parse(DeepgramFluxSTT(), AudioInputConfig(model="flux-general-en", language="es"))
+        assert "language_hint" not in q
+
+    def test_foreign_model_replaced(self) -> None:
+        # Only TIMBAL_STT_PROVIDER switched; env stt_model still Scribe's.
+        q = self._parse(DeepgramFluxSTT(), AudioInputConfig(model="scribe_v2_realtime"))
+        assert q["model"] == [DEFAULT_FLUX_MODEL]
+
+    def test_extra_passthrough_filtered(self) -> None:
+        config = AudioInputConfig(
+            extra={
+                "eot_threshold": 0.8,
+                "eot_timeout_ms": 3000,
+                "keyterm": ["Timbal", "Scribe"],
+                # Scribe-only knob must not leak into the Flux query.
+                "commit_strategy": "vad",
+            }
+        )
+        q = self._parse(DeepgramFluxSTT(), config)
+        assert q["eot_threshold"] == ["0.8"]
+        assert q["eot_timeout_ms"] == ["3000"]
+        assert q["keyterm"] == ["Timbal", "Scribe"]
+        assert "commit_strategy" not in q
+
+
+class TestNovaEventMapping:
+    def _results(self, text: str, *, is_final: bool, speech_final: bool = False, **kw) -> dict:
+        return {
+            "type": "Results",
+            "is_final": is_final,
+            "speech_final": speech_final,
+            "channel": {"alternatives": [{"transcript": text}]},
+            **kw,
+        }
+
+    async def test_interim_is_partial(self) -> None:
+        stt = DeepgramNovaSTT()
+        await stt._handle_message(self._results("hello", is_final=False))
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [("partial", "hello")]
+
+    async def test_is_final_buffers_until_speech_final(self) -> None:
+        """Deepgram's documented recipe: concat is_final segments, flush on speech_final."""
+        stt = DeepgramNovaSTT()
+        await stt._handle_message(self._results("yeah so my credit card number is two two", is_final=True))
+        assert _drain(stt) == []
+        await stt._handle_message(self._results("two two three three three three", is_final=True, speech_final=True))
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [
+            ("committed", "yeah so my credit card number is two two two two three three three three")
+        ]
+        assert stt._segments == []
+
+    async def test_partial_includes_buffered_segments(self) -> None:
+        stt = DeepgramNovaSTT()
+        await stt._handle_message(self._results("hello there", is_final=True))
+        await stt._handle_message(self._results("how are", is_final=False))
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [("partial", "hello there how are")]
+
+    async def test_from_finalize_flushes(self) -> None:
+        """commit() → Finalize → is_final response with from_finalize must commit."""
+        stt = DeepgramNovaSTT()
+        await stt._handle_message(self._results("stop the music", is_final=True, from_finalize=True))
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [("committed", "stop the music")]
+
+    async def test_utterance_end_flushes_buffer(self) -> None:
+        stt = DeepgramNovaSTT()
+        await stt._handle_message(self._results("hello there", is_final=True))
+        await stt._handle_message({"type": "UtteranceEnd", "last_word_end": 1.2})
+        events = _drain(stt)
+        assert [(e.type, e.text) for e in events] == [("committed", "hello there")]
+
+    async def test_empty_speech_final_no_commit(self) -> None:
+        stt = DeepgramNovaSTT()
+        await stt._handle_message(self._results("", is_final=True, speech_final=True))
+        assert _drain(stt) == []
+
+    async def test_error_surfaces(self) -> None:
+        stt = DeepgramNovaSTT()
+        await stt._handle_message({"type": "Error", "description": "NET-0001"})
+        events = _drain(stt)
+        assert events[0].type == "error"
+        assert "NET-0001" in events[0].text
+
+    async def test_commit_sends_finalize(self) -> None:
+        stt = DeepgramNovaSTT()
+        ws = _FakeWs()
+        stt._ws = ws
+        await stt.commit()
+        assert [json.loads(m) for m in ws.sent if isinstance(m, str)] == [{"type": "Finalize"}]
+
+
+class TestNovaUri:
+    def _parse(self, config: AudioInputConfig) -> dict:
+        uri = DeepgramNovaSTT()._build_uri(config)
+        parsed = urlparse(uri)
+        assert parsed.path == "/v1/listen"
+        return parse_qs(parsed.query)
+
+    def test_defaults(self) -> None:
+        q = self._parse(AudioInputConfig())
+        assert q["model"] == [DEFAULT_NOVA_MODEL]
+        assert q["encoding"] == ["linear16"]
+        assert q["interim_results"] == ["true"]
+        assert q["smart_format"] == ["true"]
+        assert q["punctuate"] == ["true"]
+        assert q["endpointing"] == ["300"]
+        assert q["channels"] == ["1"]
+
+    def test_language_and_extra_override(self) -> None:
+        q = self._parse(AudioInputConfig(language="es", extra={"endpointing": 500, "utterance_end_ms": 1000}))
+        assert q["language"] == ["es"]
+        assert q["endpointing"] == ["500"]
+        assert q["utterance_end_ms"] == ["1000"]
+
+    def test_foreign_model_replaced(self) -> None:
+        q = self._parse(AudioInputConfig(model="scribe_v2_realtime"))
+        assert q["model"] == [DEFAULT_NOVA_MODEL]
+        q = self._parse(AudioInputConfig(model="flux-general-en"))
+        assert q["model"] == [DEFAULT_NOVA_MODEL]
+
+    def test_nova_variant_kept(self) -> None:
+        q = self._parse(AudioInputConfig(model="nova-3-medical"))
+        assert q["model"] == ["nova-3-medical"]
+
+
+class TestResolveStt:
+    def test_explicit_providers(self) -> None:
+        assert isinstance(resolve_stt("elevenlabs"), ElevenLabsRealtimeSTT)
+        assert isinstance(resolve_stt("deepgram"), DeepgramFluxSTT)
+        assert isinstance(resolve_stt("deepgram", model="nova-3"), DeepgramNovaSTT)
+        assert isinstance(resolve_stt("deepgram", model="flux-general-en"), DeepgramFluxSTT)
+        # Leftover Scribe model + bare deepgram must NOT silently become Nova.
+        assert isinstance(resolve_stt("deepgram", model="scribe_v2_realtime"), DeepgramFluxSTT)
+
+    def test_ui_labels(self) -> None:
+        assert isinstance(resolve_stt("deepgram-flux"), DeepgramFluxSTT)
+        assert isinstance(resolve_stt("deepgram-nova"), DeepgramNovaSTT)
+        # Label wins over a stale model id from env defaults.
+        assert isinstance(resolve_stt("deepgram-flux", model="scribe_v2_realtime"), DeepgramFluxSTT)
+        assert isinstance(resolve_stt("deepgram-nova", model="flux-general-multi"), DeepgramNovaSTT)
+
+    def test_inference_from_model(self) -> None:
+        assert isinstance(resolve_stt(None, model="flux-general-multi"), DeepgramFluxSTT)
+        assert isinstance(resolve_stt(None, model="nova-3"), DeepgramNovaSTT)
+        assert isinstance(resolve_stt(None, model="scribe_v2_realtime"), ElevenLabsRealtimeSTT)
+        assert isinstance(resolve_stt(None, model=None), ElevenLabsRealtimeSTT)
+
+    def test_effective_stt_model(self) -> None:
+        assert effective_stt_model(DeepgramFluxSTT(), "scribe_v2_realtime") == DEFAULT_FLUX_MODEL
+        assert effective_stt_model(DeepgramFluxSTT(), "flux-general-en") == "flux-general-en"
+        assert effective_stt_model(DeepgramNovaSTT(), "scribe_v2_realtime") == DEFAULT_NOVA_MODEL
+        assert effective_stt_model(DeepgramNovaSTT(), "nova-3-general") == "nova-3-general"
+
+    def test_unknown_provider_raises(self) -> None:
+        with pytest.raises(ValueError, match="Unknown STT provider"):
+            resolve_stt("whisper-cpp")
+
+    def test_is_flux_model(self) -> None:
+        assert is_flux_model("flux-general-multi")
+        assert is_flux_model("FLUX-GENERAL-EN")
+        assert not is_flux_model("nova-3")
+        assert not is_flux_model(None)
+        assert not is_flux_model("")