feat(turns): let external-turn STT decide the turn start, retire provisional_vad

api/alembic/versions/c4e21b7f80a9_retire_provisional_vad_turn_start.py

@@ -0,0 +1,93 @@
+"""retire the provisional_vad turn start strategy
+
+Revision ID: c4e21b7f80a9
+Revises: f3a1c47b9e02
+Create Date: 2026-09-13 17:10:00.000000
+
+"""
+
+import json
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "c4e21b7f80a9"
+down_revision: Union[str, None] = "f3a1c47b9e02"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# `provisional_vad` gated the user turn start on a transcript, so the turn start
+# resolved from a queued frame and the interruption it broadcast flushed the
+# queued end-of-turn proposal behind it — the turn then never closed. It is
+# retired in favour of `default`, which follows the STT's own turn boundaries
+# when the STT reports them and falls back to transcript + VAD when it does not.
+#
+# `provisional_vad_pause_secs` only configured that strategy, so it goes too.
+#
+# Run history (workflow_runs and friends) records what actually executed and is
+# deliberately not rewritten. `workflow_definitions` rows are immutable
+# versions, but a run executes against the definition it points at, so they are
+# rewritten here as well — otherwise a re-run of an old definition would still
+# ask for the retired strategy.
+_RETIRED_STRATEGY = "provisional_vad"
+_REPLACEMENT_STRATEGY = "default"
+_RETIRED_KEY = "provisional_vad_pause_secs"
+
+_TARGETS = (
+    ("workflows", "workflow_configurations"),
+    ("workflow_definitions", "workflow_configurations"),
+)
+
+
+def _rewrite(config: dict) -> bool:
+    """Rewrite one configuration dict in place. True if anything changed."""
+    changed = False
+    if config.get("turn_start_strategy") == _RETIRED_STRATEGY:
+        config["turn_start_strategy"] = _REPLACEMENT_STRATEGY
+        changed = True
+    if _RETIRED_KEY in config:
+        del config[_RETIRED_KEY]
+        changed = True
+    return changed
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for table, column in _TARGETS:
+        # These are `json`, not `jsonb`, so key containment (`?`) is unavailable
+        # and a ::jsonb cast is avoided.
+        #
+        # The two operators are not interchangeable here. `->>` extracts the
+        # value as text, which is what the strategy comparison wants, but it
+        # renders a JSON null as SQL NULL — indistinguishable from a missing
+        # key. Stored configs do carry explicit nulls for keys the user never
+        # configured (see WorkflowConfigurationDefaults._treat_null_as_unset),
+        # so the key check uses `->`, which returns the JSON value and is SQL
+        # NULL only when the key is genuinely absent.
+        rows = conn.execute(
+            sa.text(
+                f"SELECT id, {column} FROM {table} "
+                f"WHERE {column} IS NOT NULL "
+                f"AND ({column}->>'turn_start_strategy' = :retired "
+                f"     OR {column}->'{_RETIRED_KEY}' IS NOT NULL)"
+            ),
+            {"retired": _RETIRED_STRATEGY},
+        ).fetchall()
+
+        for row_id, raw in rows:
+            config = json.loads(raw) if isinstance(raw, str) else raw
+            if not isinstance(config, dict) or not _rewrite(config):
+                continue
+            conn.execute(
+                sa.text(f"UPDATE {table} SET {column} = :cfg WHERE id = :id"),
+                {"cfg": json.dumps(config), "id": row_id},
+            )
+
+
+def downgrade() -> None:
+    # One-way: the retired strategy no longer exists in the application, so
+    # restoring the value would leave rows the code cannot honour. Workflows
+    # that were on it now read as `default`, which is a valid configuration.
+    pass

api/routes/telephony.py

@@ -794,7 +794,9 @@ async def _handle_telephony_websocket(
             pass
 
 
-@router.api_route("/inbound/run", methods=["GET", "POST"])
+# Exotel's inbound webhook is a GET, everyone else POSTs.
+@router.get("/inbound/run")
+@router.post("/inbound/run")
 async def handle_inbound_run(request: Request):
     """Workflow-agnostic inbound dispatcher.
 

api/schemas/workflow_configurations.py

@@ -24,7 +24,6 @@
 DEFAULT_SMART_TURN_STOP_SECS = 2.0
 DEFAULT_TURN_START_STRATEGY = "default"
 DEFAULT_TURN_START_MIN_WORDS = 3
-DEFAULT_PROVISIONAL_VAD_PAUSE_SECS = 1.5
 DEFAULT_TURN_STOP_STRATEGY = "transcription"
 DEFAULT_CONTEXT_COMPACTION_ENABLED = False
 MAX_CALL_DISPOSITIONS = 50
@@ -145,11 +144,8 @@ def _treat_null_as_unset(cls, data):
     )
     max_user_idle_timeout: float = DEFAULT_MAX_USER_IDLE_TIMEOUT_SECONDS
     smart_turn_stop_secs: float = DEFAULT_SMART_TURN_STOP_SECS
-    turn_start_strategy: Literal["default", "min_words", "provisional_vad"] = (
-        DEFAULT_TURN_START_STRATEGY
-    )
+    turn_start_strategy: Literal["default", "min_words"] = DEFAULT_TURN_START_STRATEGY
     turn_start_min_words: int = DEFAULT_TURN_START_MIN_WORDS
-    provisional_vad_pause_secs: float = DEFAULT_PROVISIONAL_VAD_PAUSE_SECS
     turn_stop_strategy: Literal["transcription", "turn_analyzer"] = (
         DEFAULT_TURN_STOP_STRATEGY
     )
@@ -177,6 +173,17 @@ def _treat_null_as_unset(cls, data):
         max_length=MAX_EXTERNAL_PBX_LEAD_HEADERS,
     )
 
+    @field_validator("turn_start_strategy", mode="before")
+    @classmethod
+    def _coerce_retired_turn_start_strategy(cls, value: object) -> object:
+        # "provisional_vad" was retired. The runtime already reads this key off
+        # the raw dict and falls through to the default for anything it does not
+        # recognise, so a row the data migration missed still runs correctly —
+        # this keeps such a row loadable (and re-savable) through the API too.
+        if value == "provisional_vad":
+            return DEFAULT_TURN_START_STRATEGY
+        return value
+
     @field_validator("call_dispositions")
     @classmethod
     def validate_call_dispositions(

api/services/pipecat/run_pipeline.py

@@ -11,7 +11,6 @@
 from api.schemas.workflow_configurations import (
     DEFAULT_MAX_CALL_DURATION_SECONDS,
     DEFAULT_MAX_USER_IDLE_TIMEOUT_SECONDS,
-    DEFAULT_PROVISIONAL_VAD_PAUSE_SECS,
     DEFAULT_SMART_TURN_STOP_SECS,
     DEFAULT_TURN_START_MIN_WORDS,
     DEFAULT_TURN_START_STRATEGY,
@@ -104,7 +103,6 @@
 from pipecat.turns.user_start import (
     ExternalUserTurnStartStrategy,
     MinWordsUserTurnStartStrategy,
-    ProvisionalVADUserTurnStartStrategy,
 )
 from pipecat.turns.user_start.transcription_user_turn_start_strategy import (
     TranscriptionUserTurnStartStrategy,
@@ -197,22 +195,20 @@ def _resolve_turn_start_min_words(run_configs: dict) -> int:
     )
 
 
-def _resolve_provisional_vad_pause_secs(run_configs: dict) -> float:
-    return max(
-        0.1,
-        float(
-            run_configs.get(
-                "provisional_vad_pause_secs", DEFAULT_PROVISIONAL_VAD_PAUSE_SECS
-            )
-        ),
-    )
-
-
 def _create_non_realtime_user_turn_start_strategies(
     run_configs: dict, *, uses_external_turns: bool
 ):
     """Return user turn start strategies for non-realtime pipelines."""
 
+    # An STT that reports its own turn boundaries decides the turn start,
+    # whatever `turn_start_strategy` asks for.
+    #
+    # Local VAD is deliberately kept out of these start strategies too: it would
+    # win the race on raw voice activity and start the turn before the STT
+    # confirms a real turn.
+    if uses_external_turns:
+        return [ExternalUserTurnStartStrategy(enable_interruptions=True)]
+
     turn_start_strategy = run_configs.get(
         "turn_start_strategy", DEFAULT_TURN_START_STRATEGY
     )
@@ -224,20 +220,6 @@ def _create_non_realtime_user_turn_start_strategies(
             )
         ]
 
-    if turn_start_strategy == "provisional_vad":
-        return [
-            ProvisionalVADUserTurnStartStrategy(
-                pause_secs=_resolve_provisional_vad_pause_secs(run_configs)
-            ),
-        ]
-
-    if uses_external_turns:
-        # The STT emits its own turn boundaries and owns interruptions. Local
-        # VAD is deliberately kept out of the default start strategies: it would
-        # win the race on raw voice activity and start the turn before the STT
-        # confirms a real turn.
-        return [ExternalUserTurnStartStrategy(enable_interruptions=True)]
-
     return [TranscriptionUserTurnStartStrategy(), VADUserTurnStartStrategy()]
 
 
@@ -1002,9 +984,12 @@ async def send_node_transition(
         turn_start_strategy = run_configs.get(
             "turn_start_strategy", DEFAULT_TURN_START_STRATEGY
         )
+        # `requested` is what the workflow asked for; `resolved` is what the
+        # pipeline built, which differs whenever external turns override it.
         logger.info(
             f"[run {workflow_run_id}] Non-realtime interrupt strategy "
             f"requested={turn_start_strategy} "
+            f"resolved={','.join(type(s).__name__ for s in user_turn_start_strategies)} "
             f"uses_external_turns={uses_external_turns}"
         )
 

api/tests/test_run_pipeline_realtime_turn_config.py

@@ -1,10 +1,7 @@
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import BotStartedSpeakingFrame, TranscriptionFrame
-from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_start import (
     ExternalUserTurnStartStrategy,
     MinWordsUserTurnStartStrategy,
-    ProvisionalVADUserTurnStartStrategy,
     TranscriptionUserTurnStartStrategy,
 )
 from pipecat.turns.user_start.vad_user_turn_start_strategy import (
@@ -19,7 +16,6 @@
 import api.services.pipecat.run_pipeline as run_pipeline_module
 from api.services.configuration.registry import ServiceProviders
 from api.services.pipecat.run_pipeline import (
-    DEFAULT_PROVISIONAL_VAD_PAUSE_SECS,
     DEFAULT_TURN_START_MIN_WORDS,
     DEFAULT_USER_TURN_STOP_TIMEOUT,
     EXTERNAL_TURN_USER_STOP_TIMEOUT,
@@ -173,70 +169,54 @@ def test_non_realtime_can_use_min_words_start_strategy():
     assert strategies[0]._min_words == 4
 
 
-def test_non_realtime_explicit_min_words_overrides_external_turn_default():
+def test_external_turn_stt_overrides_an_explicit_min_words_request():
+    """An STT that reports its own turn boundaries decides the turn start.
+
+    min_words gates the start on transcript text, which would ignore the
+    provider's turn detection on the start side while still using it to end the
+    turn, and would resolve the start from a queued frame — the shape that lets
+    a turn's own interruption flush the stop proposal queued behind it.
+    """
     strategies = _create_non_realtime_user_turn_start_strategies(
         {"turn_start_strategy": "min_words", "turn_start_min_words": 4},
         uses_external_turns=True,
     )
 
     assert len(strategies) == 1
-    assert isinstance(strategies[0], MinWordsUserTurnStartStrategy)
-    assert strategies[0]._min_words == 4
+    assert isinstance(strategies[0], ExternalUserTurnStartStrategy)
 
 
-def test_non_realtime_min_words_start_strategy_has_default_threshold():
+def test_external_turn_stt_overrides_a_retired_strategy_value():
+    """A definition still carrying "provisional_vad" resolves, it does not raise."""
     strategies = _create_non_realtime_user_turn_start_strategies(
-        {"turn_start_strategy": "min_words"},
-        uses_external_turns=False,
+        {"turn_start_strategy": "provisional_vad"},
+        uses_external_turns=True,
     )
 
     assert len(strategies) == 1
-    assert isinstance(strategies[0], MinWordsUserTurnStartStrategy)
-    assert strategies[0]._min_words == DEFAULT_TURN_START_MIN_WORDS
+    assert isinstance(strategies[0], ExternalUserTurnStartStrategy)
 
 
-def test_non_realtime_can_use_provisional_vad_start_strategy():
+def test_retired_strategy_value_falls_back_to_default_without_external_turns():
     strategies = _create_non_realtime_user_turn_start_strategies(
         {"turn_start_strategy": "provisional_vad"},
         uses_external_turns=False,
     )
 
-    assert len(strategies) == 1
-    assert isinstance(strategies[0], ProvisionalVADUserTurnStartStrategy)
-    assert strategies[0]._pause_secs == DEFAULT_PROVISIONAL_VAD_PAUSE_SECS
+    assert len(strategies) == 2
+    assert isinstance(strategies[0], TranscriptionUserTurnStartStrategy)
+    assert isinstance(strategies[1], VADUserTurnStartStrategy)
 
 
-def test_non_realtime_provisional_vad_uses_configured_pause_secs():
+def test_non_realtime_min_words_start_strategy_has_default_threshold():
     strategies = _create_non_realtime_user_turn_start_strategies(
-        {"turn_start_strategy": "provisional_vad", "provisional_vad_pause_secs": 0.4},
+        {"turn_start_strategy": "min_words"},
         uses_external_turns=False,
     )
 
     assert len(strategies) == 1
-    assert isinstance(strategies[0], ProvisionalVADUserTurnStartStrategy)
-    assert strategies[0]._pause_secs == 0.4
-
-
-async def test_non_realtime_provisional_vad_starts_on_transcript_without_vad():
-    strategies = _create_non_realtime_user_turn_start_strategies(
-        {"turn_start_strategy": "provisional_vad"},
-        uses_external_turns=False,
-    )
-    strategy = strategies[0]
-    turn_started = False
-
-    @strategy.event_handler("on_user_turn_started")
-    async def on_user_turn_started(strategy, params):
-        nonlocal turn_started
-        turn_started = True
-
-    await strategy.process_frame(BotStartedSpeakingFrame())
-    result = await strategy.process_frame(
-        TranscriptionFrame(text="Hello", user_id="user", timestamp="")
-    )
-
-    assert result == ProcessFrameResult.STOP
-    assert turn_started is True
+    assert isinstance(strategies[0], MinWordsUserTurnStartStrategy)
+    assert strategies[0]._min_words == DEFAULT_TURN_START_MIN_WORDS
 
 
 def test_non_realtime_uses_external_stop_for_external_turn_stt():

api/tests/test_workflow_configurations_schema.py

@@ -9,6 +9,7 @@
 from api.schemas.workflow_configurations import (
     DEFAULT_CALL_DISPOSITION_OPTIONS,
     DEFAULT_MAX_CALL_DURATION_SECONDS,
+    DEFAULT_TURN_START_STRATEGY,
     MAX_CALL_DISPOSITION_CODE_LENGTH,
     MAX_CALL_DISPOSITION_DESCRIPTION_LENGTH,
     MAX_CALL_DISPOSITION_DESCRIPTIONS_TOTAL_LENGTH,
@@ -106,6 +107,34 @@ def test_null_values_treated_as_unset():
     assert config.model_dump(exclude_unset=True) == {}
 
 
+def test_retired_turn_start_strategy_loads_as_default():
+    """A workflow saved before provisional_vad was retired must still load.
+
+    workflow_definitions rows are immutable versions, so one can outlive the
+    data migration (a fresh restore, a replica lagging a deploy). It has to
+    read back and re-save through the API rather than fail validation.
+    """
+    config = WorkflowConfigurationDefaults.model_validate(
+        {
+            "turn_start_strategy": "provisional_vad",
+            "provisional_vad_pause_secs": 0.4,
+        }
+    )
+
+    assert config.turn_start_strategy == DEFAULT_TURN_START_STRATEGY
+    # The retired companion key is not a field any more; extra="allow" keeps it
+    # rather than rejecting the row, and nothing reads it.
+    assert not hasattr(type(config), "provisional_vad_pause_secs")
+
+
+def test_unknown_turn_start_strategy_is_still_rejected():
+    """Coercion is scoped to the retired value, not a blanket fallback."""
+    with pytest.raises(ValidationError):
+        WorkflowConfigurationDefaults.model_validate(
+            {"turn_start_strategy": "not_a_strategy"}
+        )
+
+
 def test_call_dispositions_are_trimmed():
     configured = WorkflowConfigurationDefaults(
         call_dispositions=[

api/utils/template_renderer.py

@@ -9,7 +9,6 @@
 
 from loguru import logger
 
-
 # Regex for matching {{ variable }} template placeholders.
 # Captures: group(1) = variable path, group(2) = filter name, group(3) = filter value.
 TEMPLATE_VAR_PATTERN = r"\{\{\s*([^|\s}]+)(?:\s*\|\s*([^:}]+)(?::([^}]+))?)?\s*\}\}"