Voce Codegen

python/tests/codegen/test_set_config.py

@@ -99,6 +99,17 @@ def test_set_description(self, workspace):
         ns = _exec_agent(output)
         assert ns["agent"].description == "A helpful agent"
 
+    def test_set_voice_config(self, workspace):
+        ws = workspace("""\
+        from timbal.core import Agent
+
+        agent = Agent(name="a", model="openai/gpt-4o-mini")
+        """)
+        config = json.dumps({"voice_config": {"language": "en", "turn_detector": "local", "tts_extra": {"auto_mode": True}}})
+        output = _run_dry(ws, "--config", config)
+        ns = _exec_agent(output)
+        assert ns["agent"].voice_config == {"language": "en", "turn_detector": "local", "tts_extra": {"auto_mode": True}}
+
     def test_set_max_iter(self, workspace):
         ws = workspace("""\
         from timbal.core import Agent

python/tests/server/test_voice_config.py

@@ -12,10 +12,12 @@
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
+from pydantic import ValidationError
 from timbal import __version__ as timbal_version
 from timbal.server import voice as voice_routes
 from timbal.server.http import create_app, lifespan
 from timbal.utils import ImportSpec
+from timbal.voice.config import DEFAULT_VOICE_ID, RecordingConfig, VoiceConfig
 
 from .voice_env import VOICE_ENV_KEYS
 
@@ -24,33 +26,33 @@
 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
-        assert cfg["language"] == "es"
-        assert cfg["sample_rate"] == 16_000
-        assert cfg["stt_extra"]["commit_strategy"] == "vad"
-        assert cfg["tts_extra"]["auto_mode"] is True
+        assert cfg.stt_provider == "elevenlabs"
+        assert cfg.stt_model == "scribe_v2_realtime"
+        assert cfg.tts_model == "eleven_flash_v2_5"
+        assert cfg.voice == DEFAULT_VOICE_ID
+        assert cfg.language is None  # provider auto-detect
+        assert cfg.sample_rate == 16_000
+        assert cfg.stt_extra["commit_strategy"] == "vad"
+        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"
+        assert cfg.stt_provider == "deepgram"
+        assert cfg.stt_model == "custom_stt"
+        assert cfg.tts_model == "custom_tts"
+        assert cfg.language == "en"
 
     def test_elevenlabs_voice_id_precedence(self, monkeypatch: pytest.MonkeyPatch) -> None:
         monkeypatch.delenv("ELEVENLABS_VOICE_ID", raising=False)
         monkeypatch.delenv("TIMBAL_VOICE_ID", raising=False)
         monkeypatch.setenv("TIMBAL_VOICE_ID", "from_timbal")
-        assert voice_routes.default_voice_config_from_env()["voice"] == "from_timbal"
+        assert voice_routes.default_voice_config_from_env().voice == "from_timbal"
         monkeypatch.setenv("ELEVENLABS_VOICE_ID", "from_el")
-        assert voice_routes.default_voice_config_from_env()["voice"] == "from_el"
+        assert voice_routes.default_voice_config_from_env().voice == "from_el"
 
 
 @pytest.mark.usefixtures("clear_voice_env")
@@ -60,17 +62,17 @@ class R:
             pass
 
         merged = voice_routes.merge_voice_config(R())
-        assert merged["language"] == "es"
-        assert merged["voice"] == voice_routes._DEFAULT_VOICE_ID
+        assert merged.language is None
+        assert merged.voice == DEFAULT_VOICE_ID
 
     def test_dict_overrides_top_level(self) -> None:
         class R:
             voice_config = {"voice": "v1", "language": "pt"}
 
         merged = voice_routes.merge_voice_config(R())
-        assert merged["voice"] == "v1"
-        assert merged["language"] == "pt"
-        assert merged["stt_model"] == "scribe_v2_realtime"
+        assert merged.voice == "v1"
+        assert merged.language == "pt"
+        assert merged.stt_model == "scribe_v2_realtime"
 
     def test_callable_voice_config(self) -> None:
         class R:
@@ -79,44 +81,90 @@ def voice_config():
                 return {"voice": "callable_v"}
 
         merged = voice_routes.merge_voice_config(R())
-        assert merged["voice"] == "callable_v"
+        assert merged.voice == "callable_v"
+
+    def test_voice_config_instance(self) -> None:
+        class R:
+            voice_config = VoiceConfig(voice="typed_v", language="de")
+
+        merged = voice_routes.merge_voice_config(R())
+        assert merged.voice == "typed_v"
+        assert merged.language == "de"
+        assert merged.stt_model == "scribe_v2_realtime"
 
     def test_stt_extra_deep_merge(self) -> None:
         class R:
             voice_config = {"stt_extra": {"vad_threshold": 0.99}}
 
         merged = voice_routes.merge_voice_config(R())
-        assert merged["stt_extra"]["commit_strategy"] == "vad"
-        assert merged["stt_extra"]["vad_threshold"] == 0.99
+        assert merged.stt_extra["commit_strategy"] == "vad"
+        assert merged.stt_extra["vad_threshold"] == 0.99
 
     def test_tts_extra_deep_merge(self) -> None:
         class R:
             voice_config = {"tts_extra": {"auto_mode": False}}
 
         merged = voice_routes.merge_voice_config(R())
-        assert merged["tts_extra"]["auto_mode"] is False
+        assert merged.tts_extra["auto_mode"] is False
 
     def test_none_values_in_runnable_dict_skipped(self) -> None:
         class R:
             voice_config = {"voice": None, "language": "it"}
 
         merged = voice_routes.merge_voice_config(R())
-        assert merged["voice"] == voice_routes._DEFAULT_VOICE_ID
-        assert merged["language"] == "it"
+        assert merged.voice == DEFAULT_VOICE_ID
+        assert merged.language == "it"
+
+    def test_unknown_key_fails_fast(self) -> None:
+        """A typo'd voice_config key must raise (at server boot), not silently no-op."""
+
+        class R:
+            voice_config = {"languag": "en"}
+
+        with pytest.raises(ValidationError, match="languag"):
+            voice_routes.merge_voice_config(R())
+
+    def test_recording_dict_is_validated(self) -> None:
+        class R:
+            voice_config = {"recording": {"dir": "/tmp/rec", "layout": "sideways"}}
+
+        with pytest.raises(ValidationError, match="layout"):
+            voice_routes.merge_voice_config(R())
+
+    def test_turn_detector_instance_passes_through(self) -> None:
+        sentinel = object()
+
+        class R:
+            voice_config = {"turn_detector": sentinel}
+
+        assert voice_routes.merge_voice_config(R()).turn_detector is sentinel
 
 
 class TestMergeClientVoiceOverrides:
     def test_overlay(self) -> None:
-        base = {"voice": "a", "language": "es", "sample_rate": 16_000}
+        base = VoiceConfig(voice="a", language="es")
         out = voice_routes.merge_client_voice_overrides(base, {"language": "en", "voice": "b"})
-        assert out["language"] == "en"
-        assert out["voice"] == "b"
-        assert out["sample_rate"] == 16_000
+        assert out.language == "en"
+        assert out.voice == "b"
+        assert out.sample_rate == 16_000
 
     def test_none_skipped(self) -> None:
-        base = {"voice": "a", "language": "es"}
+        base = VoiceConfig(voice="a", language="es")
         out = voice_routes.merge_client_voice_overrides(base, {"language": None})
-        assert out["language"] == "es"
+        assert out.language == "es"
+
+    def test_non_allowlisted_keys_ignored(self) -> None:
+        base = VoiceConfig(recording=RecordingConfig(dir="/srv/rec"))
+        out = voice_routes.merge_client_voice_overrides(
+            base,
+            {"recording": {"dir": "/tmp/evil"}, "turn_detector": "heuristic", "bogus": 1, "voice": "b"},
+        )
+        assert out.recording is not None and out.recording.dir == "/srv/rec"
+        assert out.voice == "b"
+
+    def test_allowlist_is_subset_of_model_fields(self) -> None:
+        assert voice_routes.CLIENT_SETTABLE_VOICE_FIELDS <= set(VoiceConfig.model_fields)
+        assert "recording" not in voice_routes.CLIENT_SETTABLE_VOICE_FIELDS
 
 
 class TestLifespanVoiceState:
@@ -139,9 +187,9 @@ async def test_lifespan_sets_voice_config_from_runnable(
         app = FastAPI()
         spec = ImportSpec.from_fqn(os.environ["TIMBAL_RUNNABLE"])
         async with lifespan(app, spec):
-            assert app.state.voice_config["language"] == "nl"
-            assert app.state.voice_config["voice"] == "nl_voice"
-            assert "stt_extra" in app.state.voice_config
+            assert app.state.voice_config.language == "nl"
+            assert app.state.voice_config.voice == "nl_voice"
+            assert app.state.voice_config.stt_extra["commit_strategy"] == "vad"
 
     @pytest.mark.asyncio
     async def test_lifespan_merge_with_env(
@@ -159,7 +207,7 @@ async def test_lifespan_merge_with_env(
         app = FastAPI()
         spec = ImportSpec.from_fqn(os.environ["TIMBAL_RUNNABLE"])
         async with lifespan(app, spec):
-            assert app.state.voice_config["voice"] == "env_only_voice"
+            assert app.state.voice_config.voice == "env_only_voice"
 
 
 class TestCreateAppVoiceIntegration:
@@ -182,7 +230,7 @@ def test_testclient_startup_sets_voice_config(
         with TestClient(app) as client:
             r = client.get("/healthcheck")
             assert r.status_code == 204
-            assert app.state.voice_config["language"] == "sv"
+            assert app.state.voice_config.language == "sv"
 
     def test_voice_page_injects_runnable_meta(
         self,

python/tests/server/test_voice_ws.py

@@ -491,7 +491,7 @@ async def start(self, config) -> None:
         shared = _TrackingDetector()
         app = create_app()
         with TestClient(app) as client:
-            app.state.voice_config = {**(app.state.voice_config or {}), "turn_detector": shared}
+            app.state.voice_config = app.state.voice_config.model_copy(update={"turn_detector": shared})
             for _ in range(2):
                 with client.websocket_connect("/voice/ws") as ws:
                     ws.send_json({})
@@ -518,7 +518,7 @@ def _run_session(self, monkeypatch, tmp_path, hello: dict, server_td=None) -> di
         app = create_app()
         with TestClient(app) as client:
             if server_td is not None:
-                app.state.voice_config = {**(app.state.voice_config or {}), "turn_detector": server_td}
+                app.state.voice_config = app.state.voice_config.model_copy(update={"turn_detector": server_td})
             with client.websocket_connect("/voice/ws") as ws:
                 ws.send_json(hello)
                 messages = _collect_ws_messages(ws)

python/timbal/codegen/transformers/set_config.py

@@ -25,6 +25,7 @@
     "api_key",
     "model_params",  # Deprecated, kept for backward compatibility
     "skills_path",
+    "voice_config",  # dict consumed by the voice server (merge_voice_config)
 }
 
 

python/timbal/server/README.md

@@ -54,17 +54,17 @@ Whether acks were received is reported per turn in `metrics.playback_acks_receiv
 
 ### Config overrides
 
-Optional **first** text frame: a JSON object merged on top of `app.state.voice_config`, which is built at startup from environment defaults and optional `runnable.voice_config` on the loaded agent (`http` lifespan).
+Optional **first** text frame: a JSON object merged on top of `app.state.voice_config`, which is built at startup from environment defaults and optional `runnable.voice_config` on the loaded agent (`http` lifespan). Server-side `voice_config` (a dict, zero-arg callable, or `timbal.voice.VoiceConfig`) is validated strictly at startup — an unknown key fails server boot instead of being silently ignored.
 
-Only send keys you need; omitted keys keep server defaults.
+Only send keys you need; omitted keys keep server defaults. Client keys are allowlist-filtered (`CLIENT_SETTABLE_VOICE_FIELDS`): the table below plus `model` (per-session LLM override, `"provider/model"`), `turn_timeout_secs`, and `turn_timeout_fallback` (`""` disables the spoken apology). Anything else — notably `recording` — is server policy and is ignored with a log line.
 
 | Key           | Description |
 |---------------|-------------|
 | `stt_provider` | `"elevenlabs"` (default), `"deepgram-flux"`, or `"deepgram-nova"` (bare `"deepgram"` routes by `stt_model`, defaulting to Flux). Deepgram needs `DEEPGRAM_API_KEY` on the server. Flux (`/v2/listen`) does model-native end-of-turn detection: the session auto-selects the `provider` turn detector (explicit `turn_detector` still wins) and disables local VAD endpointing. Nova-3 (`/v1/listen`) is plain ASR — Timbal turn detection and VAD endpointing work exactly as with ElevenLabs. Env default: `TIMBAL_STT_PROVIDER`. |
 | `stt_model`   | Speech-to-text model id (ElevenLabs realtime `scribe_*`, Deepgram `flux-general-en`/`flux-general-multi`/`nova-3*`). Model ids that don't belong to the selected provider are ignored (provider default used). |
 | `tts_model`   | Text-to-speech model id. |
 | `voice`       | ElevenLabs voice id string. |
-| `language`    | e.g. `"es"`. |
+| `language`    | e.g. `"es"`. Unset → provider auto-detect. |
 | `sample_rate` | Hz; STT/TTS audio use this unless extended later. |
 | `encoding`    | Default `"pcm_s16le"`. |
 | `stt_extra`   | Object merged with default STT options (e.g. VAD). |

python/timbal/server/rtc.py

@@ -32,6 +32,7 @@
 from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 
+from ..voice.config import VoiceConfig
 from .voice import build_voice_session, event_to_payloads, merge_client_voice_overrides
 
 logger = structlog.get_logger("timbal.server.rtc")
@@ -92,8 +93,8 @@ 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."})
 
-    defaults: dict = getattr(request.app.state, "voice_config", None) or {}
-    sample_rate = int(merge_client_voice_overrides(defaults, config).get("sample_rate", 16_000))
+    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(