From e1b1dc16ec61d76743c1627f978bc74c8bafe881 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 12 Sep 2025 12:23:50 -0400
Subject: [PATCH 01/11] Add unit tests for `ServiceSwitcher`

---
 tests/test_service_switcher.py | 254 +++++++++++++++++++++++++++++++++
 1 file changed, 254 insertions(+)
 create mode 100644 tests/test_service_switcher.py

diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
new file mode 100644
index 000000000..9417cd2a5
--- /dev/null
+++ b/tests/test_service_switcher.py
@@ -0,0 +1,254 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Unit tests for ServiceSwitcher and related components."""
+
+import asyncio
+import unittest
+
+from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, TextFrame
+from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.tests.utils import run_test
+
+
+class MockFrameProcessor(FrameProcessor):
+    """A test frame processor that tracks which frames it has processed."""
+
+    def __init__(self, test_name: str, **kwargs):
+        """Initialize the test processor with a name.
+
+        Args:
+            test_name: A unique name for this processor instance.
+            **kwargs: Additional arguments passed to the parent FrameProcessor.
+        """
+        super().__init__(name=test_name, **kwargs)
+        self.test_name = test_name
+        self.processed_frames = []
+        self.frame_count = 0
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process an incoming frame and track it.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow in the pipeline.
+        """
+        await super().process_frame(frame, direction)
+        self.processed_frames.append(frame)
+        self.frame_count += 1
+        await self.push_frame(frame, direction)
+
+    def reset_counters(self):
+        """Reset the frame tracking counters."""
+        self.processed_frames = []
+        self.frame_count = 0
+
+
+class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
+    """Test cases for ServiceSwitcherStrategyManual."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.service1 = MockFrameProcessor("service1")
+        self.service2 = MockFrameProcessor("service2")
+        self.service3 = MockFrameProcessor("service3")
+        self.services = [self.service1, self.service2, self.service3]
+
+    def test_init_with_services(self):
+        """Test initialization with a list of services."""
+        strategy = ServiceSwitcherStrategyManual(self.services)
+
+        self.assertEqual(strategy.services, self.services)
+        self.assertEqual(strategy.active_service, self.service1)  # First service should be active
+
+    def test_init_with_empty_services(self):
+        """Test initialization with an empty list of services."""
+        strategy = ServiceSwitcherStrategyManual([])
+
+        self.assertEqual(strategy.services, [])
+        self.assertIsNone(strategy.active_service)
+
+    def test_handle_frame_manually_switch_service(self):
+        """Test manual service switching with ManuallySwitchServiceFrame."""
+        strategy = ServiceSwitcherStrategyManual(self.services)
+
+        # Initially service1 should be active
+        self.assertEqual(strategy.active_service, self.service1)
+        self.assertNotEqual(strategy.active_service, self.service2)
+
+        # Switch to service2
+        switch_frame = ManuallySwitchServiceFrame(service=self.service2)
+        strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        self.assertNotEqual(strategy.active_service, self.service1)
+        self.assertEqual(strategy.active_service, self.service2)
+        self.assertNotEqual(strategy.active_service, self.service3)
+
+        # Switch to service3
+        switch_frame = ManuallySwitchServiceFrame(service=self.service3)
+        strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        self.assertNotEqual(strategy.active_service, self.service1)
+        self.assertNotEqual(strategy.active_service, self.service2)
+        self.assertEqual(strategy.active_service, self.service3)
+
+    def test_handle_frame_invalid_service(self):
+        """Test that switching to an invalid service raises an error."""
+        strategy = ServiceSwitcherStrategyManual(self.services)
+        invalid_service = MockFrameProcessor("invalid")
+
+        switch_frame = ManuallySwitchServiceFrame(service=invalid_service)
+
+        with self.assertRaises(ValueError) as context:
+            strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        self.assertIn("Service", str(context.exception))
+        self.assertIn("is not in the list of available services", str(context.exception))
+
+    def test_handle_frame_unsupported_frame_type(self):
+        """Test that unsupported frame types raise an error."""
+        strategy = ServiceSwitcherStrategyManual(self.services)
+        unsupported_frame = TextFrame(text="test")  # Not a ServiceSwitcherFrame
+
+        with self.assertRaises(ValueError) as context:
+            strategy.handle_frame(unsupported_frame, FrameDirection.DOWNSTREAM)
+
+        self.assertIn("Unsupported frame type", str(context.exception))
+
+
+class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
+    """Test cases for ServiceSwitcher."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.service1 = MockFrameProcessor("service1")
+        self.service2 = MockFrameProcessor("service2")
+        self.service3 = MockFrameProcessor("service3")
+        self.services = [self.service1, self.service2, self.service3]
+
+    def test_init_with_manual_strategy(self):
+        """Test initialization with manual strategy."""
+        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
+
+        self.assertEqual(switcher.services, self.services)
+        self.assertIsInstance(switcher.strategy, ServiceSwitcherStrategyManual)
+        self.assertEqual(switcher.strategy.services, self.services)
+
+    async def test_default_active_service(self):
+        """Test that the initially-active service receives frames while others don't."""
+        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
+
+        # Reset counters
+        for service in self.services:
+            service.reset_counters()
+
+        # Send some test frames
+        frames_to_send = [
+            TextFrame(text="Hello 1"),
+            TextFrame(text="Hello 2"),
+            TextFrame(text="Hello 3"),
+        ]
+
+        await run_test(
+            switcher,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[TextFrame, TextFrame, TextFrame],
+        )
+
+        # Only service1 should have processed the text frames
+        # Note: The service also receives StartFrame and EndFrame, so count those too
+        text_frames = [f for f in self.service1.processed_frames if isinstance(f, TextFrame)]
+        self.assertEqual(len(text_frames), 3)
+
+        # Check that other services don't receive text frames (they might get StartFrame/EndFrame)
+        service2_text_frames = [
+            f for f in self.service2.processed_frames if isinstance(f, TextFrame)
+        ]
+        service3_text_frames = [
+            f for f in self.service3.processed_frames if isinstance(f, TextFrame)
+        ]
+        self.assertEqual(len(service2_text_frames), 0)
+        self.assertEqual(len(service3_text_frames), 0)
+
+        # Verify the actual text frames processed
+        for i, frame in enumerate(text_frames):
+            self.assertEqual(frame.text, f"Hello {i + 1}")
+
+    async def test_service_switching(self):
+        """Test that after service switching the new active service receives frames while others don't."""
+        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
+
+        # Reset counters
+        for service in self.services:
+            service.reset_counters()
+
+        # Send a test frame, a switch frame, and another test frame
+        await run_test(
+            switcher,
+            frames_to_send=[
+                TextFrame("Frame for service1"),
+                ManuallySwitchServiceFrame(service=self.service2),
+                TextFrame("Frame for service2"),
+            ],
+            expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame],
+        )
+
+        # Verify service2 received the frame
+        service1_text_frames = [
+            f for f in self.service1.processed_frames if isinstance(f, TextFrame)
+        ]
+        service2_text_frames = [
+            f for f in self.service2.processed_frames if isinstance(f, TextFrame)
+        ]
+        service3_text_frames = [
+            f for f in self.service3.processed_frames if isinstance(f, TextFrame)
+        ]
+
+        self.assertEqual(len(service1_text_frames), 1)
+        self.assertEqual(len(service2_text_frames), 1)
+        self.assertEqual(len(service3_text_frames), 0)
+        self.assertEqual(service1_text_frames[0].text, "Frame for service1")
+        self.assertEqual(service2_text_frames[0].text, "Frame for service2")
+
+    async def test_service_switching_functionality(self):
+        """Test that switching between services works correctly."""
+        # This test is simplified to avoid timing issues
+        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
+
+        # Test that we can programmatically switch services
+        self.assertEqual(switcher.strategy.active_service, self.service1)  # Initially service1
+
+        # Manually switch to service2
+        switch_frame = ManuallySwitchServiceFrame(service=self.service2)
+        switcher.strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        # Verify the switch worked
+        self.assertNotEqual(switcher.strategy.active_service, self.service1)
+        self.assertEqual(switcher.strategy.active_service, self.service2)
+        self.assertNotEqual(switcher.strategy.active_service, self.service3)
+
+        # Switch to service3
+        switch_frame = ManuallySwitchServiceFrame(service=self.service3)
+        switcher.strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        # Verify the switch worked
+        self.assertNotEqual(switcher.strategy.active_service, self.service1)
+        self.assertNotEqual(switcher.strategy.active_service, self.service2)
+        self.assertEqual(switcher.strategy.active_service, self.service3)
+
+    async def test_service_switching_with_empty_services_list(self):
+        """Test behavior with an empty services list."""
+        # ServiceSwitcher should handle empty services gracefully, but ParallelPipeline needs at least one pipeline
+        # So this test verifies that an exception is raised as expected
+        with self.assertRaises(Exception) as context:
+            ServiceSwitcher([], ServiceSwitcherStrategyManual)
+
+        self.assertIn("ParallelPipeline needs at least one argument", str(context.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

From de51637b7755b8b788afb7afc58b955fc14bc59d Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 12 Sep 2025 13:40:19 -0400
Subject: [PATCH 02/11] Update `ServiceSwitcher` so that
 `ServiceSwitcherFrame`s (which might update the currently active service) are
 processed and have an effect at the expected time. We should be able to, for
 example, queue:

- A text frame
- A `ManuallySwitchServiceFrame` (which is a `ServiceSwitcherFrame`)
- Another text frame

And expect that the first text frame be handled by the initially active service and the second text frame be handled by the newly active one.

Previously, the `ManuallySwitchServiceFrame` would have an effect too early, causing both text frames to be handled by the newly active service. Why? Because the frame filtering condition was being updated *directly* by the `ServiceSwitcher`, which is upstream from the services it's switching between. It could therefore update the filters *before* the services received the prior frames.
---
 src/pipecat/pipeline/service_switcher.py | 87 +++++++++++++++---------
 1 file changed, 56 insertions(+), 31 deletions(-)

diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index 7dd36c503..b8b1e7c41 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -6,9 +6,15 @@
 
 """Service switcher for switching between different services at runtime, with different switching strategies."""
 
+from dataclasses import dataclass
 from typing import Any, Generic, List, Optional, Type, TypeVar
 
-from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, ServiceSwitcherFrame
+from pipecat.frames.frames import (
+    ControlFrame,
+    Frame,
+    ManuallySwitchServiceFrame,
+    ServiceSwitcherFrame,
+)
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.processors.filters.function_filter import FunctionFilter
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -22,19 +28,6 @@ class ServiceSwitcherStrategy:
         self.services = services
         self.active_service: Optional[FrameProcessor] = None
 
-    def is_active(self, service: FrameProcessor) -> bool:
-        """Determine if the given service is the currently active one.
-
-        This method should be overridden by subclasses to implement specific logic.
-
-        Args:
-            service: The service to check.
-
-        Returns:
-            True if the given service is the active one, False otherwise.
-        """
-        raise NotImplementedError("Subclasses must implement this method.")
-
     def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
         """Handle a frame that controls service switching.
 
@@ -60,17 +53,6 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
         super().__init__(services)
         self.active_service = services[0] if services else None
 
-    def is_active(self, service: FrameProcessor) -> bool:
-        """Check if the given service is the currently active one.
-
-        Args:
-            service: The service to check.
-
-        Returns:
-            True if the given service is the active one, False otherwise.
-        """
-        return service == self.active_service
-
     def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
         """Handle a frame that controls service switching.
 
@@ -108,6 +90,38 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
         self.services = services
         self.strategy = strategy
 
+    class ServiceSwitcherFilter(FunctionFilter):
+        """An internal filter that allows frames to pass through to the wrapped service only if it's the active service."""
+
+        def __init__(
+            self,
+            wrapped_service: FrameProcessor,
+            active_service: FrameProcessor,
+            direction: FrameDirection,
+        ):
+            """Initialize the service switcher filter with a strategy and direction."""
+
+            async def filter(_: Frame) -> bool:
+                return self.wrapped_service == self.active_service
+
+            super().__init__(filter, direction)
+            self.wrapped_service = wrapped_service
+            self.active_service = active_service
+
+        async def process_frame(self, frame, direction):
+            """Process a frame through the filter, handling special internal filter-updating frames."""
+            if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame):
+                self.active_service = frame.active_service
+                return
+
+            await super().process_frame(frame, direction)
+
+    @dataclass
+    class ServiceSwitcherFilterFrame(ControlFrame):
+        """An internal frame used by ServiceSwitcher to filter frames based on active service."""
+
+        active_service: FrameProcessor
+
     @staticmethod
     def _make_pipeline_definitions(
         services: List[FrameProcessor], strategy: ServiceSwitcherStrategy
@@ -121,14 +135,18 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
     def _make_pipeline_definition(
         service: FrameProcessor, strategy: ServiceSwitcherStrategy
     ) -> Any:
-        async def filter(frame) -> bool:
-            _ = frame
-            return strategy.is_active(service)
-
         return [
-            FunctionFilter(filter, direction=FrameDirection.DOWNSTREAM),
+            ServiceSwitcher.ServiceSwitcherFilter(
+                wrapped_service=service,
+                active_service=strategy.active_service,
+                direction=FrameDirection.DOWNSTREAM,
+            ),
             service,
-            FunctionFilter(filter, direction=FrameDirection.UPSTREAM),
+            ServiceSwitcher.ServiceSwitcherFilter(
+                wrapped_service=service,
+                active_service=strategy.active_service,
+                direction=FrameDirection.UPSTREAM,
+            ),
         ]
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -142,3 +160,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
 
         if isinstance(frame, ServiceSwitcherFrame):
             self.strategy.handle_frame(frame, direction)
+            service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
+                active_service=self.strategy.active_service
+            )
+            # Queue frame that updates filters with new active service
+            # (Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines)
+            for p in self._pipelines:
+                await p.queue_frame(service_switcher_filter_frame, direction)

From 0839b48da836faee2b20ec8c3b8400351d4855ad Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 12 Sep 2025 14:41:54 -0400
Subject: [PATCH 03/11] Fix an issue where the upstream `ServiceSwitcherFilter`
 wouldn't get updated with the current active service

---
 src/pipecat/pipeline/service_switcher.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index b8b1e7c41..8dc7c0362 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -112,6 +112,11 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
             """Process a frame through the filter, handling special internal filter-updating frames."""
             if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame):
                 self.active_service = frame.active_service
+                # Two ServiceSwitcherFilters "sandwich" a service. Push the
+                # frame only to update the other side of the sandwich, but
+                # otherwise don't let it leave the sandwich.
+                if direction == self._direction:
+                    await self.push_frame(frame, direction)
                 return
 
             await super().process_frame(frame, direction)

From a1f84e1b508dcbcfa6821b3629a3b517f9de1f70 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 12 Sep 2025 14:42:33 -0400
Subject: [PATCH 04/11] Remove extraneous unit tests

---
 tests/test_service_switcher.py | 35 ----------------------------------
 1 file changed, 35 deletions(-)

diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
index 9417cd2a5..9ee01ea57 100644
--- a/tests/test_service_switcher.py
+++ b/tests/test_service_switcher.py
@@ -214,41 +214,6 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(service1_text_frames[0].text, "Frame for service1")
         self.assertEqual(service2_text_frames[0].text, "Frame for service2")
 
-    async def test_service_switching_functionality(self):
-        """Test that switching between services works correctly."""
-        # This test is simplified to avoid timing issues
-        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
-
-        # Test that we can programmatically switch services
-        self.assertEqual(switcher.strategy.active_service, self.service1)  # Initially service1
-
-        # Manually switch to service2
-        switch_frame = ManuallySwitchServiceFrame(service=self.service2)
-        switcher.strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
-
-        # Verify the switch worked
-        self.assertNotEqual(switcher.strategy.active_service, self.service1)
-        self.assertEqual(switcher.strategy.active_service, self.service2)
-        self.assertNotEqual(switcher.strategy.active_service, self.service3)
-
-        # Switch to service3
-        switch_frame = ManuallySwitchServiceFrame(service=self.service3)
-        switcher.strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
-
-        # Verify the switch worked
-        self.assertNotEqual(switcher.strategy.active_service, self.service1)
-        self.assertNotEqual(switcher.strategy.active_service, self.service2)
-        self.assertEqual(switcher.strategy.active_service, self.service3)
-
-    async def test_service_switching_with_empty_services_list(self):
-        """Test behavior with an empty services list."""
-        # ServiceSwitcher should handle empty services gracefully, but ParallelPipeline needs at least one pipeline
-        # So this test verifies that an exception is raised as expected
-        with self.assertRaises(Exception) as context:
-            ServiceSwitcher([], ServiceSwitcherStrategyManual)
-
-        self.assertIn("ParallelPipeline needs at least one argument", str(context.exception))
-
 
 if __name__ == "__main__":
     unittest.main()

From b814b70e1e2e63bbc53ed88b64682ae99be6c8af Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Fri, 12 Sep 2025 15:36:56 -0400
Subject: [PATCH 05/11] Allow controlling `ServiceSwitcher` with either
 immediate frames (`SystemFrame`s) or in-order frames (`ControlFrame`s).

Immediate is the "default", i.e. has the more obvious name (e.g. `ManuallySwitchServiceFrame` v `ManuallySwitchServiceControlFrame`), since that's *probably* what users will want to reach for. Also, the immediate frames are more likely to behave like what we had before the last few commits, where the service switch would always "jump the queue" by having an immediate effect once it hit the `ServiceSwitcher` in the pipeline, jumping ahead of frames in front of it destined for the service.
---
 src/pipecat/frames/frames.py             | 39 ++++++++++--
 src/pipecat/pipeline/service_switcher.py | 47 ++++++++++----
 tests/test_service_switcher.py           | 78 ++++++++++++++++++++++--
 3 files changed, 142 insertions(+), 22 deletions(-)

diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index 012ab9af7..a9b58d986 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -1603,17 +1603,46 @@ class MixerEnableFrame(MixerControlFrame):
 
 
 @dataclass
-class ServiceSwitcherFrame(ControlFrame):
-    """A base class for frames that control ServiceSwitcher behavior."""
+class ManuallySwitchServiceBaseFrame:
+    """Base class for frames that request a manual service switch from a ServiceSwitcher.
+
+    Must be inherited alongside either ServiceSwitcherFrame or ServiceSwitcherControlFrame.
+    """
+
+    service: "FrameProcessor"
+
+
+@dataclass
+class ServiceSwitcherFrame(SystemFrame):
+    """A base class for frames that affect ServiceSwitcher behavior immediately."""
 
     pass
 
 
 @dataclass
-class ManuallySwitchServiceFrame(ServiceSwitcherFrame):
-    """A frame to request a manual switch in the active service in a ServiceSwitcher.
+class ManuallySwitchServiceFrame(ServiceSwitcherFrame, ManuallySwitchServiceBaseFrame):
+    """A frame to request an immediate manual switch in the active service in a ServiceSwitcher.
 
     Handled by ServiceSwitcherStrategyManual to switch the active service.
     """
 
-    service: "FrameProcessor"
+    pass
+
+
+@dataclass
+class ServiceSwitcherControlFrame(ControlFrame):
+    """A base class for frames that affect ServiceSwitcher behavior, in order."""
+
+    pass
+
+
+@dataclass
+class ManuallySwitchServiceControlFrame(
+    ServiceSwitcherControlFrame, ManuallySwitchServiceBaseFrame
+):
+    """A frame to request a manual switch in the active service in a ServiceSwitcher, in order.
+
+    Handled by ServiceSwitcherStrategyManual to switch the active service.
+    """
+
+    pass
diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index 8dc7c0362..41efcb974 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -12,8 +12,10 @@ from typing import Any, Generic, List, Optional, Type, TypeVar
 from pipecat.frames.frames import (
     ControlFrame,
     Frame,
-    ManuallySwitchServiceFrame,
+    ManuallySwitchServiceBaseFrame,
+    ServiceSwitcherControlFrame,
     ServiceSwitcherFrame,
+    SystemFrame,
 )
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.processors.filters.function_filter import FunctionFilter
@@ -28,7 +30,9 @@ class ServiceSwitcherStrategy:
         self.services = services
         self.active_service: Optional[FrameProcessor] = None
 
-    def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
+    def handle_frame(
+        self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection
+    ):
         """Handle a frame that controls service switching.
 
         This method can be overridden by subclasses to implement specific logic
@@ -53,14 +57,16 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
         super().__init__(services)
         self.active_service = services[0] if services else None
 
-    def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
+    def handle_frame(
+        self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection
+    ):
         """Handle a frame that controls service switching.
 
         Args:
             frame: The frame to handle.
             direction: The direction of the frame (upstream or downstream).
         """
-        if isinstance(frame, ManuallySwitchServiceFrame):
+        if isinstance(frame, ManuallySwitchServiceBaseFrame):
             self._set_active(frame.service)
         else:
             raise ValueError(f"Unsupported frame type: {type(frame)}")
@@ -110,7 +116,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
 
         async def process_frame(self, frame, direction):
             """Process a frame through the filter, handling special internal filter-updating frames."""
-            if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame):
+            if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterBaseFrame):
                 self.active_service = frame.active_service
                 # Two ServiceSwitcherFilters "sandwich" a service. Push the
                 # frame only to update the other side of the sandwich, but
@@ -122,11 +128,23 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
             await super().process_frame(frame, direction)
 
     @dataclass
-    class ServiceSwitcherFilterFrame(ControlFrame):
-        """An internal frame used by ServiceSwitcher to filter frames based on active service."""
+    class ServiceSwitcherFilterBaseFrame:
+        """Base class for internal frames used by ServiceSwitcher to filter frames based on active service."""
 
         active_service: FrameProcessor
 
+    @dataclass
+    class ServiceSwitcherFilterFrame(SystemFrame, ServiceSwitcherFilterBaseFrame):
+        """An internal frame used by ServiceSwitcher to filter frames based on active service."""
+
+        pass
+
+    @dataclass
+    class ServiceSwitcherFilterControlFrame(ControlFrame, ServiceSwitcherFilterBaseFrame):
+        """An internal control frame used by ServiceSwitcher to filter frames based on active service."""
+
+        pass
+
     @staticmethod
     def _make_pipeline_definitions(
         services: List[FrameProcessor], strategy: ServiceSwitcherStrategy
@@ -163,11 +181,18 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, ServiceSwitcherFrame):
+        if isinstance(frame, (ServiceSwitcherFrame, ServiceSwitcherControlFrame)):
             self.strategy.handle_frame(frame, direction)
-            service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
-                active_service=self.strategy.active_service
-            )
+            if isinstance(frame, ServiceSwitcherFrame):
+                # Apply immediate update (system frame)
+                service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
+                    active_service=self.strategy.active_service
+                )
+            else:
+                # Apply in-order update (control frame)
+                service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterControlFrame(
+                    active_service=self.strategy.active_service
+                )
             # Queue frame that updates filters with new active service
             # (Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines)
             for p in self._pipelines:
diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
index 9ee01ea57..b8a508c82 100644
--- a/tests/test_service_switcher.py
+++ b/tests/test_service_switcher.py
@@ -9,7 +9,12 @@
 import asyncio
 import unittest
 
-from pipecat.frames.frames import Frame, ManuallySwitchServiceFrame, TextFrame
+from pipecat.frames.frames import (
+    Frame,
+    ManuallySwitchServiceControlFrame,
+    ManuallySwitchServiceFrame,
+    TextFrame,
+)
 from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.tests.utils import run_test
@@ -72,7 +77,7 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(strategy.services, [])
         self.assertIsNone(strategy.active_service)
 
-    def test_handle_frame_manually_switch_service(self):
+    def test_handle_manually_switch_service_frame(self):
         """Test manual service switching with ManuallySwitchServiceFrame."""
         strategy = ServiceSwitcherStrategyManual(self.services)
 
@@ -96,6 +101,30 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
         self.assertNotEqual(strategy.active_service, self.service2)
         self.assertEqual(strategy.active_service, self.service3)
 
+    def test_handle_manually_switch_service_control_frame(self):
+        """Test manual service switching with ManuallySwitchServiceControlFrame."""
+        strategy = ServiceSwitcherStrategyManual(self.services)
+
+        # Initially service1 should be active
+        self.assertEqual(strategy.active_service, self.service1)
+        self.assertNotEqual(strategy.active_service, self.service2)
+
+        # Switch to service2
+        switch_frame = ManuallySwitchServiceControlFrame(service=self.service2)
+        strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        self.assertNotEqual(strategy.active_service, self.service1)
+        self.assertEqual(strategy.active_service, self.service2)
+        self.assertNotEqual(strategy.active_service, self.service3)
+
+        # Switch to service3
+        switch_frame = ManuallySwitchServiceControlFrame(service=self.service3)
+        strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
+
+        self.assertNotEqual(strategy.active_service, self.service1)
+        self.assertNotEqual(strategy.active_service, self.service2)
+        self.assertEqual(strategy.active_service, self.service3)
+
     def test_handle_frame_invalid_service(self):
         """Test that switching to an invalid service raises an error."""
         strategy = ServiceSwitcherStrategyManual(self.services)
@@ -178,8 +207,8 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         for i, frame in enumerate(text_frames):
             self.assertEqual(frame.text, f"Hello {i + 1}")
 
-    async def test_service_switching(self):
-        """Test that after service switching the new active service receives frames while others don't."""
+    async def test_service_switching_in_order(self):
+        """Test that after service switching using ManuallySwitchServiceControlFrame, the new active service receives frames while others don't."""
         switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
 
         # Reset counters
@@ -191,10 +220,10 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
             switcher,
             frames_to_send=[
                 TextFrame("Frame for service1"),
-                ManuallySwitchServiceFrame(service=self.service2),
+                ManuallySwitchServiceControlFrame(service=self.service2),
                 TextFrame("Frame for service2"),
             ],
-            expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame],
+            expected_down_frames=[TextFrame, ManuallySwitchServiceControlFrame, TextFrame],
         )
 
         # Verify service2 received the frame
@@ -214,6 +243,43 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(service1_text_frames[0].text, "Frame for service1")
         self.assertEqual(service2_text_frames[0].text, "Frame for service2")
 
+    async def test_service_switching_immediate(self):
+        """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't."""
+        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
+
+        # Reset counters
+        for service in self.services:
+            service.reset_counters()
+
+        # Send a test frame, a switch frame, and another test frame
+        await run_test(
+            switcher,
+            frames_to_send=[
+                # Out of order on purpose - ManuallySwitchServiceFrame should jump the queue
+                TextFrame("Frame 1 for service2"),
+                ManuallySwitchServiceFrame(service=self.service2),
+                TextFrame("Frame 2 for service2"),
+            ],
+            expected_down_frames=[ManuallySwitchServiceFrame, TextFrame, TextFrame],
+        )
+
+        # Verify service2 received the frame
+        service1_text_frames = [
+            f for f in self.service1.processed_frames if isinstance(f, TextFrame)
+        ]
+        service2_text_frames = [
+            f for f in self.service2.processed_frames if isinstance(f, TextFrame)
+        ]
+        service3_text_frames = [
+            f for f in self.service3.processed_frames if isinstance(f, TextFrame)
+        ]
+
+        self.assertEqual(len(service1_text_frames), 0)
+        self.assertEqual(len(service2_text_frames), 2)
+        self.assertEqual(len(service3_text_frames), 0)
+        self.assertEqual(service2_text_frames[0].text, "Frame 1 for service2")
+        self.assertEqual(service2_text_frames[1].text, "Frame 2 for service2")
+
 
 if __name__ == "__main__":
     unittest.main()

From a12392182c88ca9e2e294c675eb48651aab389f3 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Tue, 16 Sep 2025 14:32:03 -0400
Subject: [PATCH 06/11] Simplify, undoing the change allowing controlling
 `ServiceSwitcher` with immediate frames (`SystemFrame`s). Service switcher
 frames are `ControlFrame`s, which are easier to reason about. We can always
 build the immediate option later if needed (i.e. if there's sufficient user
 pull for it)

---
 src/pipecat/frames/frames.py             | 49 +++------------
 src/pipecat/pipeline/service_switcher.py | 50 ++++-----------
 tests/test_service_switcher.py           | 78 +++---------------------
 3 files changed, 30 insertions(+), 147 deletions(-)

diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index a9b58d986..ffef7c99c 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -1603,46 +1603,17 @@ class MixerEnableFrame(MixerControlFrame):
 
 
 @dataclass
-class ManuallySwitchServiceBaseFrame:
-    """Base class for frames that request a manual service switch from a ServiceSwitcher.
+class ServiceSwitcherFrame(ControlFrame):
+    """A base class for frames that affect ServiceSwitcher behavior."""
 
-    Must be inherited alongside either ServiceSwitcherFrame or ServiceSwitcherControlFrame.
+    pass
+
+
+@dataclass
+class ManuallySwitchServiceFrame(ServiceSwitcherFrame):
+    """A frame to request a manual switch in the active service in a ServiceSwitcher.
+
+    Handled by ServiceSwitcherStrategyManual to switch the active service.
     """
 
     service: "FrameProcessor"
-
-
-@dataclass
-class ServiceSwitcherFrame(SystemFrame):
-    """A base class for frames that affect ServiceSwitcher behavior immediately."""
-
-    pass
-
-
-@dataclass
-class ManuallySwitchServiceFrame(ServiceSwitcherFrame, ManuallySwitchServiceBaseFrame):
-    """A frame to request an immediate manual switch in the active service in a ServiceSwitcher.
-
-    Handled by ServiceSwitcherStrategyManual to switch the active service.
-    """
-
-    pass
-
-
-@dataclass
-class ServiceSwitcherControlFrame(ControlFrame):
-    """A base class for frames that affect ServiceSwitcher behavior, in order."""
-
-    pass
-
-
-@dataclass
-class ManuallySwitchServiceControlFrame(
-    ServiceSwitcherControlFrame, ManuallySwitchServiceBaseFrame
-):
-    """A frame to request a manual switch in the active service in a ServiceSwitcher, in order.
-
-    Handled by ServiceSwitcherStrategyManual to switch the active service.
-    """
-
-    pass
diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index 41efcb974..fe99751e4 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -12,10 +12,8 @@ from typing import Any, Generic, List, Optional, Type, TypeVar
 from pipecat.frames.frames import (
     ControlFrame,
     Frame,
-    ManuallySwitchServiceBaseFrame,
-    ServiceSwitcherControlFrame,
+    ManuallySwitchServiceFrame,
     ServiceSwitcherFrame,
-    SystemFrame,
 )
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 from pipecat.processors.filters.function_filter import FunctionFilter
@@ -30,9 +28,7 @@ class ServiceSwitcherStrategy:
         self.services = services
         self.active_service: Optional[FrameProcessor] = None
 
-    def handle_frame(
-        self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection
-    ):
+    def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
         """Handle a frame that controls service switching.
 
         This method can be overridden by subclasses to implement specific logic
@@ -57,16 +53,14 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
         super().__init__(services)
         self.active_service = services[0] if services else None
 
-    def handle_frame(
-        self, frame: ServiceSwitcherFrame | ServiceSwitcherControlFrame, direction: FrameDirection
-    ):
+    def handle_frame(self, frame: ServiceSwitcherFrame, direction: FrameDirection):
         """Handle a frame that controls service switching.
 
         Args:
             frame: The frame to handle.
             direction: The direction of the frame (upstream or downstream).
         """
-        if isinstance(frame, ManuallySwitchServiceBaseFrame):
+        if isinstance(frame, ManuallySwitchServiceFrame):
             self._set_active(frame.service)
         else:
             raise ValueError(f"Unsupported frame type: {type(frame)}")
@@ -116,7 +110,7 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
 
         async def process_frame(self, frame, direction):
             """Process a frame through the filter, handling special internal filter-updating frames."""
-            if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterBaseFrame):
+            if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame):
                 self.active_service = frame.active_service
                 # Two ServiceSwitcherFilters "sandwich" a service. Push the
                 # frame only to update the other side of the sandwich, but
@@ -128,22 +122,10 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
             await super().process_frame(frame, direction)
 
     @dataclass
-    class ServiceSwitcherFilterBaseFrame:
-        """Base class for internal frames used by ServiceSwitcher to filter frames based on active service."""
-
-        active_service: FrameProcessor
-
-    @dataclass
-    class ServiceSwitcherFilterFrame(SystemFrame, ServiceSwitcherFilterBaseFrame):
+    class ServiceSwitcherFilterFrame(ControlFrame):
         """An internal frame used by ServiceSwitcher to filter frames based on active service."""
 
-        pass
-
-    @dataclass
-    class ServiceSwitcherFilterControlFrame(ControlFrame, ServiceSwitcherFilterBaseFrame):
-        """An internal control frame used by ServiceSwitcher to filter frames based on active service."""
-
-        pass
+        active_service: FrameProcessor
 
     @staticmethod
     def _make_pipeline_definitions(
@@ -181,19 +163,11 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, (ServiceSwitcherFrame, ServiceSwitcherControlFrame)):
+        if isinstance(frame, ServiceSwitcherFrame):
             self.strategy.handle_frame(frame, direction)
-            if isinstance(frame, ServiceSwitcherFrame):
-                # Apply immediate update (system frame)
-                service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
-                    active_service=self.strategy.active_service
-                )
-            else:
-                # Apply in-order update (control frame)
-                service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterControlFrame(
-                    active_service=self.strategy.active_service
-                )
-            # Queue frame that updates filters with new active service
-            # (Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines)
+            service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
+                active_service=self.strategy.active_service
+            )
+            # Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines
             for p in self._pipelines:
                 await p.queue_frame(service_switcher_filter_frame, direction)
diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
index b8a508c82..26f6614a4 100644
--- a/tests/test_service_switcher.py
+++ b/tests/test_service_switcher.py
@@ -6,12 +6,10 @@
 
 """Unit tests for ServiceSwitcher and related components."""
 
-import asyncio
 import unittest
 
 from pipecat.frames.frames import (
     Frame,
-    ManuallySwitchServiceControlFrame,
     ManuallySwitchServiceFrame,
     TextFrame,
 )
@@ -101,30 +99,6 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
         self.assertNotEqual(strategy.active_service, self.service2)
         self.assertEqual(strategy.active_service, self.service3)
 
-    def test_handle_manually_switch_service_control_frame(self):
-        """Test manual service switching with ManuallySwitchServiceControlFrame."""
-        strategy = ServiceSwitcherStrategyManual(self.services)
-
-        # Initially service1 should be active
-        self.assertEqual(strategy.active_service, self.service1)
-        self.assertNotEqual(strategy.active_service, self.service2)
-
-        # Switch to service2
-        switch_frame = ManuallySwitchServiceControlFrame(service=self.service2)
-        strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
-
-        self.assertNotEqual(strategy.active_service, self.service1)
-        self.assertEqual(strategy.active_service, self.service2)
-        self.assertNotEqual(strategy.active_service, self.service3)
-
-        # Switch to service3
-        switch_frame = ManuallySwitchServiceControlFrame(service=self.service3)
-        strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
-
-        self.assertNotEqual(strategy.active_service, self.service1)
-        self.assertNotEqual(strategy.active_service, self.service2)
-        self.assertEqual(strategy.active_service, self.service3)
-
     def test_handle_frame_invalid_service(self):
         """Test that switching to an invalid service raises an error."""
         strategy = ServiceSwitcherStrategyManual(self.services)
@@ -207,8 +181,8 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         for i, frame in enumerate(text_frames):
             self.assertEqual(frame.text, f"Hello {i + 1}")
 
-    async def test_service_switching_in_order(self):
-        """Test that after service switching using ManuallySwitchServiceControlFrame, the new active service receives frames while others don't."""
+    async def test_service_switching(self):
+        """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't."""
         switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
 
         # Reset counters
@@ -219,11 +193,11 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         await run_test(
             switcher,
             frames_to_send=[
-                TextFrame("Frame for service1"),
-                ManuallySwitchServiceControlFrame(service=self.service2),
-                TextFrame("Frame for service2"),
+                TextFrame("Hello 1"),
+                ManuallySwitchServiceFrame(service=self.service2),
+                TextFrame("Hello 2"),
             ],
-            expected_down_frames=[TextFrame, ManuallySwitchServiceControlFrame, TextFrame],
+            expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame],
         )
 
         # Verify service2 received the frame
@@ -240,45 +214,9 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(len(service1_text_frames), 1)
         self.assertEqual(len(service2_text_frames), 1)
         self.assertEqual(len(service3_text_frames), 0)
-        self.assertEqual(service1_text_frames[0].text, "Frame for service1")
-        self.assertEqual(service2_text_frames[0].text, "Frame for service2")
 
-    async def test_service_switching_immediate(self):
-        """Test that after service switching using ManuallySwitchServiceFrame, the new active service receives frames while others don't."""
-        switcher = ServiceSwitcher(self.services, ServiceSwitcherStrategyManual)
-
-        # Reset counters
-        for service in self.services:
-            service.reset_counters()
-
-        # Send a test frame, a switch frame, and another test frame
-        await run_test(
-            switcher,
-            frames_to_send=[
-                # Out of order on purpose - ManuallySwitchServiceFrame should jump the queue
-                TextFrame("Frame 1 for service2"),
-                ManuallySwitchServiceFrame(service=self.service2),
-                TextFrame("Frame 2 for service2"),
-            ],
-            expected_down_frames=[ManuallySwitchServiceFrame, TextFrame, TextFrame],
-        )
-
-        # Verify service2 received the frame
-        service1_text_frames = [
-            f for f in self.service1.processed_frames if isinstance(f, TextFrame)
-        ]
-        service2_text_frames = [
-            f for f in self.service2.processed_frames if isinstance(f, TextFrame)
-        ]
-        service3_text_frames = [
-            f for f in self.service3.processed_frames if isinstance(f, TextFrame)
-        ]
-
-        self.assertEqual(len(service1_text_frames), 0)
-        self.assertEqual(len(service2_text_frames), 2)
-        self.assertEqual(len(service3_text_frames), 0)
-        self.assertEqual(service2_text_frames[0].text, "Frame 1 for service2")
-        self.assertEqual(service2_text_frames[1].text, "Frame 2 for service2")
+        self.assertEqual(service1_text_frames[0].text, "Hello 1")
+        self.assertEqual(service2_text_frames[0].text, "Hello 2")
 
 
 if __name__ == "__main__":

From 5b55988846233812e12bf6f5557acdff0c7327ef Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Wed, 17 Sep 2025 15:38:28 -0400
Subject: [PATCH 07/11] Denote a couple of variables are private with a leading
 underscore

---
 src/pipecat/pipeline/service_switcher.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index fe99751e4..c079bdab4 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -102,16 +102,16 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
             """Initialize the service switcher filter with a strategy and direction."""
 
             async def filter(_: Frame) -> bool:
-                return self.wrapped_service == self.active_service
+                return self._wrapped_service == self._active_service
 
             super().__init__(filter, direction)
-            self.wrapped_service = wrapped_service
-            self.active_service = active_service
+            self._wrapped_service = wrapped_service
+            self._active_service = active_service
 
         async def process_frame(self, frame, direction):
             """Process a frame through the filter, handling special internal filter-updating frames."""
             if isinstance(frame, ServiceSwitcher.ServiceSwitcherFilterFrame):
-                self.active_service = frame.active_service
+                self._active_service = frame.active_service
                 # Two ServiceSwitcherFilters "sandwich" a service. Push the
                 # frame only to update the other side of the sandwich, but
                 # otherwise don't let it leave the sandwich.

From 2cd2567a37a66acf070d77bda102a32cedd6eb16 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Wed, 17 Sep 2025 16:04:30 -0400
Subject: [PATCH 08/11] Add a unit tests validating that multiple
 `ServiceSwitcher`s can be used in the same pipeline (currently failing)

---
 tests/test_service_switcher.py | 93 ++++++++++++++++++++++++++++++++++
 1 file changed, 93 insertions(+)

diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
index 26f6614a4..b69d1f023 100644
--- a/tests/test_service_switcher.py
+++ b/tests/test_service_switcher.py
@@ -13,6 +13,7 @@ from pipecat.frames.frames import (
     ManuallySwitchServiceFrame,
     TextFrame,
 )
+from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyManual
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.tests.utils import run_test
@@ -160,6 +161,7 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
             switcher,
             frames_to_send=frames_to_send,
             expected_down_frames=[TextFrame, TextFrame, TextFrame],
+            expected_up_frames=[],  # Expect no error frames
         )
 
         # Only service1 should have processed the text frames
@@ -198,6 +200,7 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
                 TextFrame("Hello 2"),
             ],
             expected_down_frames=[TextFrame, ManuallySwitchServiceFrame, TextFrame],
+            expected_up_frames=[],  # Expect no error frames
         )
 
         # Verify service2 received the frame
@@ -218,6 +221,96 @@ class TestServiceSwitcher(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(service1_text_frames[0].text, "Hello 1")
         self.assertEqual(service2_text_frames[0].text, "Hello 2")
 
+    async def test_multi_service_switcher_targeting(self):
+        """Test that ManuallySwitchServiceFrame targets the correct ServiceSwitcher in a multi-switcher pipeline."""
+        # Create services for first switcher
+        switcher1_service1 = MockFrameProcessor("switcher1_service1")
+        switcher1_service2 = MockFrameProcessor("switcher1_service2")
+        switcher1_services = [switcher1_service1, switcher1_service2]
+
+        # Create services for second switcher
+        switcher2_service1 = MockFrameProcessor("switcher2_service1")
+        switcher2_service2 = MockFrameProcessor("switcher2_service2")
+        switcher2_services = [switcher2_service1, switcher2_service2]
+
+        # Create two service switchers
+        switcher1 = ServiceSwitcher(switcher1_services, ServiceSwitcherStrategyManual)
+        switcher2 = ServiceSwitcher(switcher2_services, ServiceSwitcherStrategyManual)
+
+        # Create a pipeline with both switchers: switcher1 -> switcher2
+        pipeline = Pipeline([switcher1, switcher2])
+
+        # Reset counters
+        for service in switcher1_services + switcher2_services:
+            service.reset_counters()
+
+        # Initially, both switchers should use their first services
+        self.assertEqual(switcher1.strategy.active_service, switcher1_service1)
+        self.assertEqual(switcher2.strategy.active_service, switcher2_service1)
+
+        # Send frames to test the pipeline:
+        # 1. Text frame (should go through both switchers' active services)
+        # 2. Switch frame targeting switcher1's second service
+        # 3. Text frame (should go through switcher1's new service and switcher2's original service)
+        # 4. Switch frame targeting switcher2's second service
+        # 5. Text frame (should go through switcher1's current service and switcher2's new service)
+        await run_test(
+            pipeline,
+            frames_to_send=[
+                TextFrame("Before any switches"),
+                ManuallySwitchServiceFrame(service=switcher1_service2),  # Switch first switcher
+                TextFrame("After switching first switcher"),
+                ManuallySwitchServiceFrame(service=switcher2_service2),  # Switch second switcher
+                TextFrame("After switching second switcher"),
+            ],
+            expected_down_frames=[
+                TextFrame,
+                ManuallySwitchServiceFrame,
+                TextFrame,
+                ManuallySwitchServiceFrame,
+                TextFrame,
+            ],
+            expected_up_frames=[],  # Expect no error frames
+        )
+
+        # Verify the active services changed correctly
+        self.assertEqual(switcher1.strategy.active_service, switcher1_service2)
+        self.assertEqual(switcher2.strategy.active_service, switcher2_service2)
+
+        # Verify frame distribution:
+        # First text frame should go through switcher1_service1 and switcher2_service1
+        switcher1_service1_texts = [
+            f for f in switcher1_service1.processed_frames if isinstance(f, TextFrame)
+        ]
+        switcher2_service1_texts = [
+            f for f in switcher2_service1.processed_frames if isinstance(f, TextFrame)
+        ]
+
+        # Second text frame should go through switcher1_service2 and switcher2_service1
+        switcher1_service2_texts = [
+            f for f in switcher1_service2.processed_frames if isinstance(f, TextFrame)
+        ]
+
+        # Third text frame should go through switcher1_service2 and switcher2_service2
+        switcher2_service2_texts = [
+            f for f in switcher2_service2.processed_frames if isinstance(f, TextFrame)
+        ]
+
+        # Verify frame counts and content
+        self.assertEqual(len(switcher1_service1_texts), 1)
+        self.assertEqual(switcher1_service1_texts[0].text, "Before any switches")
+
+        self.assertEqual(len(switcher1_service2_texts), 2)
+        self.assertEqual(switcher1_service2_texts[0].text, "After switching first switcher")
+        self.assertEqual(switcher1_service2_texts[1].text, "After switching second switcher")
+
+        self.assertEqual(len(switcher2_service1_texts), 2)
+        self.assertEqual(switcher2_service1_texts[0].text, "Before any switches")
+        self.assertEqual(switcher2_service1_texts[1].text, "After switching first switcher")
+
+        self.assertEqual(len(switcher2_service2_texts), 1)
+        self.assertEqual(switcher2_service2_texts[0].text, "After switching second switcher")
+
 
 if __name__ == "__main__":
     unittest.main()

From 8bc3c8914093fe22a6208938fb06d999f0162eec Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Wed, 17 Sep 2025 16:09:18 -0400
Subject: [PATCH 09/11] Fix a bug preventing usage of multiple
 `ServiceSwitcher`s in a pipeline

---
 src/pipecat/pipeline/service_switcher.py | 11 ++++++-----
 tests/test_service_switcher.py           | 13 -------------
 2 files changed, 6 insertions(+), 18 deletions(-)

diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index c079bdab4..985f59b3a 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -61,20 +61,21 @@ class ServiceSwitcherStrategyManual(ServiceSwitcherStrategy):
             direction: The direction of the frame (upstream or downstream).
         """
         if isinstance(frame, ManuallySwitchServiceFrame):
-            self._set_active(frame.service)
+            self._set_active_if_available(frame.service)
         else:
             raise ValueError(f"Unsupported frame type: {type(frame)}")
 
-    def _set_active(self, service: FrameProcessor):
-        """Set the active service to the given one.
+    def _set_active_if_available(self, service: FrameProcessor):
+        """Set the active service to the given one, if it is in the list of available services.
+
+        If it's not in the list, the request is ignored, as it may have been
+        intended for another ServiceSwitcher in the pipeline.
 
         Args:
             service: The service to set as active.
         """
         if service in self.services:
             self.active_service = service
-        else:
-            raise ValueError(f"Service {service} is not in the list of available services.")
 
 
 StrategyType = TypeVar("StrategyType", bound=ServiceSwitcherStrategy)
diff --git a/tests/test_service_switcher.py b/tests/test_service_switcher.py
index b69d1f023..bf80d842e 100644
--- a/tests/test_service_switcher.py
+++ b/tests/test_service_switcher.py
@@ -100,19 +100,6 @@ class TestServiceSwitcherStrategyManual(unittest.IsolatedAsyncioTestCase):
         self.assertNotEqual(strategy.active_service, self.service2)
         self.assertEqual(strategy.active_service, self.service3)
 
-    def test_handle_frame_invalid_service(self):
-        """Test that switching to an invalid service raises an error."""
-        strategy = ServiceSwitcherStrategyManual(self.services)
-        invalid_service = MockFrameProcessor("invalid")
-
-        switch_frame = ManuallySwitchServiceFrame(service=invalid_service)
-
-        with self.assertRaises(ValueError) as context:
-            strategy.handle_frame(switch_frame, FrameDirection.DOWNSTREAM)
-
-        self.assertIn("Service", str(context.exception))
-        self.assertIn("is not in the list of available services", str(context.exception))
-
     def test_handle_frame_unsupported_frame_type(self):
         """Test that unsupported frame types raise an error."""
         strategy = ServiceSwitcherStrategyManual(self.services)

From bd760deff284d3eafa5d742d0b57dffb7272a842 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Wed, 17 Sep 2025 16:19:31 -0400
Subject: [PATCH 10/11] Update comment with more detail for posterity

---
 src/pipecat/pipeline/service_switcher.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/src/pipecat/pipeline/service_switcher.py b/src/pipecat/pipeline/service_switcher.py
index 985f59b3a..64920e0ce 100644
--- a/src/pipecat/pipeline/service_switcher.py
+++ b/src/pipecat/pipeline/service_switcher.py
@@ -169,6 +169,11 @@ class ServiceSwitcher(ParallelPipeline, Generic[StrategyType]):
             service_switcher_filter_frame = ServiceSwitcher.ServiceSwitcherFilterFrame(
                 active_service=self.strategy.active_service
             )
-            # Hack: we need access ParallelPipeline internals here to queue the frame in each of the pipelines
+            # Hack: we need access ParallelPipeline internals here to queue the
+            # frame in each of the pipelines.
+            # Why not just call super().process_frame(service_switcher_filter_frame, direction),
+            # you ask? Because that would also send this internal-only frame
+            # down the "main" pipeline instead of only down the individual
+            # branches of the parallel pipeline, which we want to avoid.
             for p in self._pipelines:
                 await p.queue_frame(service_switcher_filter_frame, direction)

From 27f1e9dd6984b7006597a66d2172f804bd6679d3 Mon Sep 17 00:00:00 2001
From: Paul Kompfner <paul@daily.co>
Date: Wed, 17 Sep 2025 16:27:12 -0400
Subject: [PATCH 11/11] Update CHANGELOG with a description of the
 recently-fixed `ServiceSwitcher` bugs

---
 CHANGELOG.md | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 617b5ad49..82a5feb0b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -44,6 +44,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a couple of bugs in `ServiceSwitcher`:
+  - Using multiple `ServiceSwitcher`s in a pipeline would result in an error.
+  - `ServiceSwitcherFrame`s (such as `ManuallySwitchServiceFrame`s) were having
+    an effect too early, essentially "jumping the queue" in terms of pipeline
+    frame ordering.
+
 - Fixed a self-cancellation deadlock in `UserIdleProcessor` when returning
   `False` from an idle callback. The task now terminates naturally instead of
   attempting to cancel itself.