Repository: pipecat-ai/pipecat (python)

Issue:
Intermittent WebSocket "no close frame received or sent" errors when switching STT/LLM services via ServiceSwitcher during long calls

### pipecat version

0.0.106

### Python version

3.13

### Operating System

Docker

### Issue description

During a long-running call (~6 minutes), the pipeline intermittently crashes with WebSocket-related errors when using ServiceSwitcher to dynamically switch between STT and LLM providers.

The issue appears after several minutes into the call (around timestamp 4:12), and affects multiple services simultaneously:
- Soniox STT
- Deepgram STT (after switching)
- OpenAI LLM
- Gemini LLM (after switching)
- ElevenLabs TTS

All of them report similar WebSocket errors:
"no close frame received or sent"

This suggests a possible issue in how connections are handled, closed, or re-used when switching services mid-stream.

I previously encountered a similar issue here:
https://github.com/pipecat-ai/pipecat/issues/3913

However, in that case I was using SileroVADAnalyzer, while in this case I am using an external VAD analyzer.

- LangSmith trace:
  https://smith.langchain.com/public/2ff17901-a7a1-4304-b64b-7ebcd00a560e/r

- Call recording attached (~6 minutes)
[call_audio.wav](https://github.com/user-attachments/files/26247005/call_audio.wav)
  Issue starts around: 4:12

- messages.json attached (user + assistant aggregators)
[messages.json](https://github.com/user-attachments/files/26246985/messages.json)




```python
# STT configuration:
SonioxSTTService(
    api_key=...,
    settings=SonioxSTTSettings(
        language_hints=languages,
        language_hints_strict=language_hints_strict,
        context=context,
        model="stt-rt-v4",
    ),
    vad_force_turn_endpoint=False  # Using Soniox VAD
)

DeepgramSTTService(
    api_key=...,
    settings=DeepgramSTTSettings(
        model=model,
        language=language,
        interim_results=True,
        smart_format=True,
        punctuate=True,
        profanity_filter=False,
        vad_events=True,  # Using Deepgram VAD
    ),
)

# Aggregator configuration:

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        vad_analyzer=None  # External VAD (not Silero)
    )
)
```
- Important note:
  This issue occurs WITHOUT using SileroVADAnalyzer, unlike the previous issue.




### Reproduction steps

1. Start a real-time call using Pipecat pipeline.
2. Configure ServiceSwitcher for:
   - STT: Soniox -> Deepgram
   - LLM: OpenAI -> Gemini
3. Use external VAD analyzer (NOT SileroVADAnalyzer).
4. Use Soniox with:
   - vad_force_turn_endpoint=False (Soniox VAD)
5. Use Deepgram with:
   - vad_events=True (Deepgram VAD)
6. Let the conversation run for several minutes (~4+ minutes).
7. During the conversation:
   - ServiceSwitcher switches from Soniox to Deepgram
   - ServiceSwitcher switches from OpenAI to Gemini


### Expected behavior

- ServiceSwitcher should handle switching between STT and LLM providers seamlessly.
- WebSocket connections should be properly closed and re-established when switching services.
- Long-running calls should remain stable without cascading failures across services.

### Actual behavior

- Multiple services throw WebSocket errors:
  "Unknown error occurred: no close frame received or sent"
- Errors propagate across:
  - STT services (Soniox, Deepgram)
  - LLM services (OpenAI, Gemini)
  - TTS service (ElevenLabs)
- Pipeline enters unstable state and fails to continue processing correctly.

### Logs

```shell
WARNING  | pipecat.pipeline.service_switcher:handle_error:196 - Service SonioxSTTService#2 reported an error: Unknown error occurred: no close frame received or sent
WARNING  | pipecat.pipeline.task:_source_push_frame:903 - PipelineTask#2: Something went wrong: ErrorFrame#0(error: Unknown error occurred: no close frame received or sent
WARNING  | pipecat.services.websocket_service:_maybe_try_reconnect:149 - ElevenLabsTTSService#2 connection closed, but with an error: no close frame received or sent
ERROR    | pipecat.processors.frame_processor:push_error_frame:744 - ElevenLabsTTSService#2 exception (/usr/local/lib/python3.13/site-packages/websockets/asyncio/connection.py:957): Unknown error occurred: no close frame received or sent
WARNING  | pipecat.pipeline.service_switcher:handle_error:196 - Service OpenAILLMService#4 reported an error: Unknown error occurred: no close frame received or sent
```


Patch:
diff --git a/changelog/4149.fixed.md b/changelog/4149.fixed.md
new file mode 100644
index 0000000000..db41676330
--- /dev/null
+++ b/changelog/4149.fixed.md
@@ -0,0 +1,1 @@
+- Fixed `ServiceSwitcher` with `ServiceSwitcherStrategyFailover` incorrectly triggering failover when `ErrorFrame`s from other pipeline stages (e.g. TTS) propagated upstream through the switcher. Previously, any non-fatal error passing through would be misattributed to the active service and trigger an unwanted service switch. Now only errors originating from the switcher's own managed services trigger failover.
diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index 76b7036815..9d2e2a56ec 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -193,7 +193,8 @@ async def handle_error(self, error: ErrorFrame) -> Optional[FrameProcessor]:
             The newly active service if a switch occurred, or None if no
             other service is available.
         """
-        logger.warning(f"Service {self._active_service.name} reported an error: {error.error}")
+        service_name = error.processor.name if error.processor else self._active_service.name
+        logger.warning(f"Service {service_name} reported an error: {error.error}")
 
         if len(self._services) <= 1:
             logger.error("No other service available to switch to")
@@ -312,9 +313,11 @@ async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirect
             if frame.service_name != self.strategy.active_service.name:
                 return
 
-        # Let the strategy react to non-fatal errors from the active service.
+        # Let the strategy react to non-fatal errors from the active service,
+        # ignoring errors just propagating upstream from other processors.
         if isinstance(frame, ErrorFrame) and not frame.fatal:
-            await self.strategy.handle_error(frame)
+            if frame.processor and frame.processor == self.strategy.active_service:
+                await self.strategy.handle_error(frame)
 
         await super().push_frame(frame, direction)
 


Test patch:
diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
index 4cdf38f1a6..0ccea6e20b 100644
--- a/tests/test_service_switcher.py
+++ b/tests/test_service_switcher.py
@@ -105,6 +105,38 @@ def reset_counters(self):
         self.metadata_push_count = 0
 
 
+class ErrorInjectorProcessor(FrameProcessor):
+    """A downstream processor that pushes an ErrorFrame upstream on receiving a TextFrame.
+
+    Simulates an error from a service outside the ServiceSwitcher (e.g. TTS
+    erroring while propagating upstream through an LLM switcher).
+    """
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, TextFrame) and direction == FrameDirection.DOWNSTREAM:
+            await self.push_error("downstream service error")
+        await self.push_frame(frame, direction)
+
+
+class ErrorOnTextService(FrameProcessor):
+    """A mock service that pushes an error on the first TextFrame it receives.
+
+    Simulates a managed service inside a ServiceSwitcher that encounters an error.
+    """
+
+    def __init__(self, test_name: str, **kwargs):
+        super().__init__(name=test_name, **kwargs)
+        self._errored = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, TextFrame) and not self._errored:
+            self._errored = True
+            await self.push_error("service connection lost")
+        await self.push_frame(frame, direction)
+
+
 @dataclass
 class DummySystemFrame(SystemFrame):
     """A dummy system frame for testing purposes."""
@@ -580,6 +612,49 @@ async def test_manual_switch_still_works(self):
         self.assertEqual(result, self.service3)
         self.assertEqual(strategy.active_service, self.service3)
 
+    async def test_passthrough_error_does_not_trigger_failover(self):
+        """Test that an error propagating upstream from a downstream processor does not trigger failover.
+
+        This reproduces the bug where an ErrorFrame from e.g. TTS propagates
+        upstream through an LLM ServiceSwitcher and incorrectly triggers
+        failover even though neither LLM service produced the error.
+        """
+        switcher = ServiceSwitcher(
+            [self.service1, self.service2],
+            strategy_type=ServiceSwitcherStrategyFailover,
+        )
+        error_injector = ErrorInjectorProcessor()
+        pipeline = Pipeline([switcher, error_injector])
+
+        await run_test(
+            pipeline,
+            frames_to_send=[TextFrame(text="test")],
+            expected_down_frames=[TextFrame],
+            expected_up_frames=[ErrorFrame],
+        )
+
+        # Active service should NOT have changed — the error came from outside
+        self.assertEqual(switcher.strategy.active_service, self.service1)
+
+    async def test_managed_service_error_triggers_failover(self):
+        """Test that an error from a managed service inside the switcher triggers failover."""
+        error_service = ErrorOnTextService("error_service")
+        backup_service = MockFrameProcessor("backup_service")
+        switcher = ServiceSwitcher(
+            [error_service, backup_service],
+            strategy_type=ServiceSwitcherStrategyFailover,
+        )
+
+        await run_test(
+            switcher,
+            frames_to_send=[TextFrame(text="test")],
+            expected_down_frames=[TextFrame],
+            expected_up_frames=[ErrorFrame],
+        )
+
+        # Active service SHOULD have changed — the error came from a managed service
+        self.assertEqual(switcher.strategy.active_service, backup_service)
+
     async def test_on_service_switched_event_fires_on_error(self):
         """Test that on_service_switched event fires when an error triggers a switch."""
         strategy = ServiceSwitcherStrategyFailover(self.services)