fix(cartesia): narrow aligned_transcript when timestamps unsupported (#6493)

livekit-plugins/livekit-plugins-cartesia/livekit/plugins/cartesia/tts.py

@@ -59,6 +59,20 @@
     _is_sonic_3,
 )
 
+# Word timestamps on non-preview sonic models are limited to these languages.
+# Preview models support all languages. See Cartesia TTS docs / #6493.
+_WORD_TIMESTAMP_LANGUAGES = frozenset({"en", "de", "es", "fr"})
+
+
+def _supports_word_timestamps(model: str, language: LanguageCode | None) -> bool:
+    """Return whether Cartesia can deliver word timestamps for this config."""
+    if "preview" in model:
+        return True
+    if language is None:
+        # language unset — don't assume unsupported; API may still deliver
+        return True
+    return language.language in _WORD_TIMESTAMP_LANGUAGES
+
 
 @dataclass
 class _TTSOptions:
@@ -127,10 +141,13 @@ def __init__(
             base_url (str, optional): The base URL for the Cartesia API. Defaults to "https://api.cartesia.ai".
         """  # noqa: E501
 
+        language_code = LanguageCode(language) if language else None
+        aligned_transcript = word_timestamps and _supports_word_timestamps(model, language_code)
+
         super().__init__(
             capabilities=tts.TTSCapabilities(
                 streaming=True,
-                aligned_transcript=word_timestamps,
+                aligned_transcript=aligned_transcript,
             ),
             sample_rate=sample_rate,
             num_channels=1,
@@ -147,7 +164,7 @@ def __init__(
 
         self._opts = _TTSOptions(
             model=model,
-            language=LanguageCode(language) if language else None,
+            language=language_code,
             encoding=encoding,
             sample_rate=sample_rate,
             voice=voice,
@@ -181,22 +198,7 @@ def __init__(
         elif isinstance(text_pacing, tts.SentenceStreamPacer):
             self._stream_pacer = text_pacing
 
-        if word_timestamps:
-            if "preview" not in self._opts.model and (
-                self._opts.language is not None
-                and self._opts.language.language
-                not in {
-                    "en",
-                    "de",
-                    "es",
-                    "fr",
-                }
-            ):
-                # https://docs.cartesia.ai/api-reference/tts/compare-tts-endpoints
-                logger.warning(
-                    "word_timestamps is only supported for languages en, de, es, and fr with `sonic` models"
-                    " or all languages with `preview` models"
-                )
+        self._warn_if_aligned_transcript_unsupported()
 
     class Markup(tts.TTS.Markup):
         # markup delegation lives in the base class, keyed on _provider_key()
@@ -290,6 +292,30 @@ def update_options(
         if speed or emotion or volume or pronunciation_dict_id:
             self._check_generation_config()
 
+        # model/language changes can enable or disable timestamp delivery (#6493)
+        self._refresh_aligned_transcript_capability()
+
+    def _refresh_aligned_transcript_capability(self) -> None:
+        """Narrow ``capabilities.aligned_transcript`` to configs that can deliver timestamps."""
+        supported = _supports_word_timestamps(self._opts.model, self._opts.language)
+        self._capabilities.aligned_transcript = self._opts.word_timestamps and supported
+        self._warn_if_aligned_transcript_unsupported()
+
+    def _warn_if_aligned_transcript_unsupported(self) -> None:
+        if not self._opts.word_timestamps:
+            return
+        if _supports_word_timestamps(self._opts.model, self._opts.language):
+            return
+        # https://docs.cartesia.ai/api-reference/tts/compare-tts-endpoints
+        logger.warning(
+            "model configuration does not support aligned transcript "
+            "(word_timestamps); disabling capabilities.aligned_transcript. "
+            "Supported: languages en, de, es, and fr with `sonic` models, "
+            "or all languages with `preview` models. model=%s language=%s",
+            self._opts.model,
+            self._opts.language.language if self._opts.language else None,
+        )
+
     def synthesize(
         self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
     ) -> ChunkedStream:
@@ -603,6 +629,9 @@ def _to_cartesia_options(opts: _TTSOptions, *, streaming: bool) -> dict[str, Any
             options["generation_config"] = generation_config
 
     if streaming:
-        options["add_timestamps"] = opts.word_timestamps
+        # only request timestamps when the model/language combo can deliver them
+        options["add_timestamps"] = opts.word_timestamps and _supports_word_timestamps(
+            opts.model, opts.language
+        )
 
     return options

tests/test_plugin_cartesia_tts.py

@@ -0,0 +1,95 @@
+"""Unit tests for Cartesia TTS aligned-transcript capability narrowing (#6493)."""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from livekit.agents import LanguageCode
+from livekit.plugins.cartesia import tts as cartesia_tts
+
+pytestmark = pytest.mark.plugin("cartesia")
+
+
+def test_aligned_transcript_enabled_for_supported_language() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="en", word_timestamps=True)
+    assert tts.capabilities.aligned_transcript is True
+
+
+def test_aligned_transcript_disabled_for_unsupported_language() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="ja", word_timestamps=True)
+    assert tts.capabilities.aligned_transcript is False
+
+
+def test_aligned_transcript_disabled_when_word_timestamps_off() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="en", word_timestamps=False)
+    assert tts.capabilities.aligned_transcript is False
+
+
+def test_aligned_transcript_allowed_for_preview_model_any_language() -> None:
+    tts = cartesia_tts.TTS(
+        api_key="test-key",
+        model="sonic-preview",
+        language="ja",
+        word_timestamps=True,
+    )
+    assert tts.capabilities.aligned_transcript is True
+
+
+def test_unsupported_config_logs_warning(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level(logging.WARNING, logger="livekit.plugins.cartesia"):
+        tts = cartesia_tts.TTS(api_key="test-key", language="ja", word_timestamps=True)
+
+    assert tts.capabilities.aligned_transcript is False
+    assert any(
+        "does not support aligned transcript" in record.message for record in caplog.records
+    )
+
+
+def test_update_options_narrows_capability_when_language_becomes_unsupported() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="en", word_timestamps=True)
+    assert tts.capabilities.aligned_transcript is True
+
+    tts.update_options(language="ja")
+    assert tts.capabilities.aligned_transcript is False
+
+
+def test_update_options_restores_capability_when_language_becomes_supported() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="ja", word_timestamps=True)
+    assert tts.capabilities.aligned_transcript is False
+
+    tts.update_options(language="fr")
+    assert tts.capabilities.aligned_transcript is True
+
+
+def test_update_options_to_preview_model_enables_capability() -> None:
+    tts = cartesia_tts.TTS(
+        api_key="test-key",
+        model="sonic-3",
+        language="ja",
+        word_timestamps=True,
+    )
+    assert tts.capabilities.aligned_transcript is False
+
+    tts.update_options(model="sonic-preview")
+    assert tts.capabilities.aligned_transcript is True
+
+
+def test_add_timestamps_omitted_when_unsupported() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="ja", word_timestamps=True)
+    options = cartesia_tts._to_cartesia_options(tts._opts, streaming=True)
+    assert options["add_timestamps"] is False
+
+
+def test_add_timestamps_requested_when_supported() -> None:
+    tts = cartesia_tts.TTS(api_key="test-key", language="en", word_timestamps=True)
+    options = cartesia_tts._to_cartesia_options(tts._opts, streaming=True)
+    assert options["add_timestamps"] is True
+
+
+def test_supports_word_timestamps_helper() -> None:
+    assert cartesia_tts._supports_word_timestamps("sonic-3", LanguageCode("en")) is True
+    assert cartesia_tts._supports_word_timestamps("sonic-3", LanguageCode("ja")) is False
+    assert cartesia_tts._supports_word_timestamps("sonic-preview", LanguageCode("ja")) is True
+    assert cartesia_tts._supports_word_timestamps("sonic-3", None) is True