diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8b2b3ea03..6f962d33b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Allow passing custom pipeline sink and source processors to a
+  `Pipeline`. Pipeline source and sink processors are used to know and control
+  what's coming in and out of a `Pipeline` processor.
+
+- Added `FrameProcessor.pause_processing_system_frames()` and
+  `FrameProcessor.resume_processing_system_frames()`. These allow to pause and
+  resume the processing of system frame.
+
+- Added new `on_process_frame()` observer method which makes it possible to know
+  when a frame is being processed.
+
+- Added new `FrameProcessor.entry_processor()` method. This allows you to access
+  the first non-compound processor in a pipeline.
+
+- Added `FrameProcessor` properties `processors`, `next` and `previous`.
+
 - `ElevenLabsTTSService` now supports additional runtime changes to the `model`,
   `language`, and `voice_settings` parameters.
 
@@ -23,6 +39,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `retry_timeout_secs` and `retry_on_timeout`. This feature is disabled by
   default.
 
+### Removed
+
+- Removed unused `FrameProcessor.set_parent()` and
+  `FrameProcessor.get_parent()`.
+
 ### Fixed
 
 - Fixed an `AudioBufferProcessor` issues that would cause audio overlap when
@@ -31,6 +52,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed an issue where `AsyncAITTSService` had very high latency in responding
   by adding `force=true` when sending the flush command.
 
+### Performance
+
+- Improve `PipelineTask` performance by using direct mode processors and by
+  removing unnecessary tasks.
+
+- Improve `ParallelPipeline` performance by using direct mode, by not
+  creating a task for each frame and every sub-pipeline and also by removing
+  other unnecessary tasks.
+
+- `Pipeline` performance improvements by using direct mode.
+
 ### Other
 
 - Added `14w-function-calling-mistal.py` using `MistralLLMService`.
diff --git a/src/pipecat/observers/base_observer.py b/src/pipecat/observers/base_observer.py
index d8a67a373..a5c31b81c 100644
--- a/src/pipecat/observers/base_observer.py
+++ b/src/pipecat/observers/base_observer.py
@@ -11,7 +11,6 @@ processors without modifying the pipeline structure. Observers can be used
 for logging, debugging, analytics, and monitoring pipeline behavior.
 """
 
-from abc import abstractmethod
 from dataclasses import dataclass
 
 from typing_extensions import TYPE_CHECKING
@@ -23,6 +22,28 @@ if TYPE_CHECKING:
     from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
+@dataclass
+class FrameProcessed:
+    """Event data for frame processing in the pipeline.
+
+    Represents an event where a frame is being processed by a processor. This
+    data structure is typically used by observers to track the flow of frames
+    through the pipeline for logging, debugging, or analytics purposes.
+
+    Parameters:
+        processor: The processor processing the frame.
+        frame: The frame being processed.
+        direction: The direction of the frame (e.g., downstream or upstream).
+        timestamp: The time when the frame was pushed, based on the pipeline clock.
+
+    """
+
+    processor: "FrameProcessor"
+    frame: Frame
+    direction: "FrameDirection"
+    timestamp: int
+
+
 @dataclass
 class FramePushed:
     """Event data for frame transfers between processors in the pipeline.
@@ -56,7 +77,18 @@ class BaseObserver(BaseObject):
     performance analysis, and analytics collection.
     """
 
-    @abstractmethod
+    async def on_process_frame(self, data: FrameProcessed):
+        """Handle the event when a frame is being processed by a processor.
+
+        This method should be implemented by subclasses to define specific
+        behavior (e.g., logging, monitoring, debugging) when a frame is
+        being processed by a processor.
+
+        Args:
+            data: The event data containing details about the frame processing.
+        """
+        pass
+
     async def on_push_frame(self, data: FramePushed):
         """Handle the event when a frame is pushed from one processor to another.
 
diff --git a/src/pipecat/pipeline/base_pipeline.py b/src/pipecat/pipeline/base_pipeline.py
index a3b1e0b3e..8bc9da524 100644
--- a/src/pipecat/pipeline/base_pipeline.py
+++ b/src/pipecat/pipeline/base_pipeline.py
@@ -6,31 +6,12 @@
 
 """Base pipeline implementation for frame processing."""
 
-from abc import abstractmethod
-from typing import List
-
 from pipecat.processors.frame_processor import FrameProcessor
 
 
 class BasePipeline(FrameProcessor):
-    """Base class for all pipeline implementations.
+    """Base class for all pipeline implementations."""
 
-    Provides the foundation for pipeline processors that need to support
-    metrics collection from their contained processors.
-    """
-
-    def __init__(self):
+    def __init__(self, **kwargs):
         """Initialize the base pipeline."""
-        super().__init__()
-
-    @abstractmethod
-    def processors_with_metrics(self) -> List[FrameProcessor]:
-        """Return processors that can generate metrics.
-
-        Implementing classes should collect and return all processors within
-        their pipeline that support metrics generation.
-
-        Returns:
-            List of frame processors that support metrics collection.
-        """
-        pass
+        super().__init__(**kwargs)
diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py
index cb66f62b5..148387de7 100644
--- a/src/pipecat/pipeline/parallel_pipeline.py
+++ b/src/pipecat/pipeline/parallel_pipeline.py
@@ -11,106 +11,15 @@ sub-pipelines concurrently, with coordination for system frames and proper
 handling of pipeline lifecycle events.
 """
 
-import asyncio
 from itertools import chain
-from typing import Awaitable, Callable, Dict, List
+from typing import Dict, List
 
 from loguru import logger
 
-from pipecat.frames.frames import (
-    CancelFrame,
-    EndFrame,
-    Frame,
-    StartFrame,
-    StartInterruptionFrame,
-    SystemFrame,
-)
+from pipecat.frames.frames import EndFrame, Frame, StartFrame
 from pipecat.pipeline.base_pipeline import BasePipeline
-from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
-from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
-
-
-class ParallelPipelineSource(FrameProcessor):
-    """Source processor for parallel pipeline branches.
-
-    Handles frame routing for parallel pipeline inputs, directing system frames
-    to the parent push function and other upstream frames to a queue for processing.
-    """
-
-    def __init__(
-        self,
-        upstream_queue: asyncio.Queue,
-        push_frame_func: Callable[[Frame, FrameDirection], Awaitable[None]],
-    ):
-        """Initialize the parallel pipeline source.
-
-        Args:
-            upstream_queue: Queue for collecting upstream frames from this branch.
-            push_frame_func: Function to push frames to the parent parallel pipeline.
-        """
-        super().__init__(enable_direct_mode=True)
-        self._up_queue = upstream_queue
-        self._push_frame_func = push_frame_func
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames with special handling for system frames.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame flow.
-        """
-        await super().process_frame(frame, direction)
-
-        match direction:
-            case FrameDirection.UPSTREAM:
-                if isinstance(frame, SystemFrame):
-                    await self._push_frame_func(frame, direction)
-                else:
-                    await self._up_queue.put(frame)
-            case FrameDirection.DOWNSTREAM:
-                await self.push_frame(frame, direction)
-
-
-class ParallelPipelineSink(FrameProcessor):
-    """Sink processor for parallel pipeline branches.
-
-    Handles frame routing for parallel pipeline outputs, directing system frames
-    to the parent push function and other downstream frames to a queue for coordination.
-    """
-
-    def __init__(
-        self,
-        downstream_queue: asyncio.Queue,
-        push_frame_func: Callable[[Frame, FrameDirection], Awaitable[None]],
-    ):
-        """Initialize the parallel pipeline sink.
-
-        Args:
-            downstream_queue: Queue for collecting downstream frames from this branch.
-            push_frame_func: Function to push frames to the parent parallel pipeline.
-        """
-        super().__init__(enable_direct_mode=True)
-        self._down_queue = downstream_queue
-        self._push_frame_func = push_frame_func
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames with special handling for system frames.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame flow.
-        """
-        await super().process_frame(frame, direction)
-
-        match direction:
-            case FrameDirection.UPSTREAM:
-                await self.push_frame(frame, direction)
-            case FrameDirection.DOWNSTREAM:
-                if isinstance(frame, SystemFrame):
-                    await self._push_frame_func(frame, direction)
-                else:
-                    await self._down_queue.put(frame)
 
 
 class ParallelPipeline(BasePipeline):
@@ -132,28 +41,69 @@ class ParallelPipeline(BasePipeline):
             Exception: If no processor lists are provided.
             TypeError: If any argument is not a list of processors.
         """
+        # We don't set it to direct mode because we use frame pausing and that
+        # requires queues.
         super().__init__()
 
         if len(args) == 0:
             raise Exception(f"ParallelPipeline needs at least one argument")
 
-        self._args = args
-        self._sources = []
-        self._sinks = []
         self._pipelines = []
 
         self._seen_ids = set()
-        self._endframe_counter: Dict[int, int] = {}
-        self._start_frame_counter: Dict[int, int] = {}
-        self._started = False
+        self._frame_counter: Dict[int, int] = {}
 
-        self._up_task = None
-        self._down_task = None
+        logger.debug(f"Creating {self} pipelines")
+
+        for processors in args:
+            if not isinstance(processors, list):
+                raise TypeError(f"ParallelPipeline argument {processors} is not a list")
+
+            num_pipelines = len(self._pipelines)
+
+            # We add a source before the pipeline and a sink after so we control
+            # the frames that are pushed upstream and downstream.
+            source = PipelineSource(
+                self._parallel_push_frame, name=f"{self}::Source{num_pipelines}"
+            )
+            sink = PipelineSink(self._pipeline_sink_push_frame, name=f"{self}::Sink{num_pipelines}")
+
+            # Create pipeline
+            pipeline = Pipeline(processors, source=source, sink=sink)
+            self._pipelines.append(pipeline)
+
+        logger.debug(f"Finished creating {self} pipelines")
 
     #
-    # BasePipeline
+    # Frame processor
     #
 
+    @property
+    def processors(self):
+        """Return the list of sub-processors contained within this processor.
+
+        Only compound processors (e.g. pipelines and parallel pipelines) have
+        sub-processors. Non-compound processors will return an empty list.
+
+        Returns:
+            The list of sub-processors if this is a compound processor.
+        """
+        return self._pipelines
+
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return self._pipelines
+
     def processors_with_metrics(self) -> List[FrameProcessor]:
         """Collect processors that can generate metrics from all parallel branches.
 
@@ -162,10 +112,6 @@ class ParallelPipeline(BasePipeline):
         """
         return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines))
 
-    #
-    # Frame processor
-    #
-
     async def setup(self, setup: FrameProcessorSetup):
         """Set up the parallel pipeline and all its branches.
 
@@ -176,39 +122,14 @@ class ParallelPipeline(BasePipeline):
             TypeError: If any processor list argument is not actually a list.
         """
         await super().setup(setup)
-
-        self._up_queue = WatchdogQueue(setup.task_manager)
-        self._down_queue = WatchdogQueue(setup.task_manager)
-
-        logger.debug(f"Creating {self} pipelines")
-        for processors in self._args:
-            if not isinstance(processors, list):
-                raise TypeError(f"ParallelPipeline argument {processors} is not a list")
-
-            # We will add a source before the pipeline and a sink after.
-            source = ParallelPipelineSource(self._up_queue, self._parallel_push_frame)
-            sink = ParallelPipelineSink(self._down_queue, self._pipeline_sink_push_frame)
-            self._sources.append(source)
-            self._sinks.append(sink)
-
-            # Create pipeline
-            pipeline = Pipeline(processors)
-            source.link(pipeline)
-            pipeline.link(sink)
-            self._pipelines.append(pipeline)
-
-        logger.debug(f"Finished creating {self} pipelines")
-
-        await asyncio.gather(*[s.setup(setup) for s in self._sources])
-        await asyncio.gather(*[p.setup(setup) for p in self._pipelines])
-        await asyncio.gather(*[s.setup(setup) for s in self._sinks])
+        for p in self._pipelines:
+            await p.setup(setup)
 
     async def cleanup(self):
         """Clean up the parallel pipeline and all its branches."""
         await super().cleanup()
-        await asyncio.gather(*[s.cleanup() for s in self._sources])
-        await asyncio.gather(*[p.cleanup() for p in self._pipelines])
-        await asyncio.gather(*[s.cleanup() for s in self._sinks])
+        for p in self._pipelines:
+            await p.cleanup()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames through all parallel branches with lifecycle coordination.
@@ -219,79 +140,15 @@ class ParallelPipeline(BasePipeline):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, StartFrame):
-            self._start_frame_counter[frame.id] = len(self._pipelines)
-        elif isinstance(frame, EndFrame):
-            self._endframe_counter[frame.id] = len(self._pipelines)
-        elif isinstance(frame, CancelFrame):
-            await self._cancel()
+        # Parallel pipeline synchronized frames.
+        if isinstance(frame, (StartFrame, EndFrame)):
+            self._frame_counter[frame.id] = len(self._pipelines)
+            await self.pause_processing_system_frames()
+            await self.pause_processing_frames()
 
-        if direction == FrameDirection.UPSTREAM:
-            # If we get an upstream frame we process it in each sink.
-            await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks])
-        elif direction == FrameDirection.DOWNSTREAM:
-            # If we get a downstream frame we process it in each source.
-            await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources])
-
-        # Handle interruptions after everything has been cancelled.
-        if isinstance(frame, StartInterruptionFrame):
-            await self._handle_interruption()
-        # Wait for tasks to finish.
-        elif isinstance(frame, EndFrame):
-            await self._stop()
-
-    async def _start(self, frame: StartFrame):
-        """Start the parallel pipeline processing tasks."""
-        await self._create_tasks()
-
-    async def _stop(self):
-        """Stop all parallel pipeline processing tasks."""
-        if self._up_task:
-            # The up task doesn't receive an EndFrame, so we just cancel it.
-            await self.cancel_task(self._up_task)
-            self._up_task = None
-
-        if self._down_task:
-            # The down tasks waits for the last EndFrame sent by the internal
-            # pipelines.
-            await self._down_task
-            self._down_task = None
-
-    async def _cancel(self):
-        """Cancel all parallel pipeline processing tasks."""
-        if self._up_task:
-            self._up_queue.cancel()
-            await self.cancel_task(self._up_task)
-            self._up_task = None
-        if self._down_task:
-            self._down_queue.cancel()
-            await self.cancel_task(self._down_task)
-            self._down_task = None
-
-    async def _create_tasks(self):
-        """Create upstream and downstream processing tasks if not already running."""
-        if not self._up_task:
-            self._up_task = self.create_task(self._process_up_queue())
-        if not self._down_task:
-            self._down_task = self.create_task(self._process_down_queue())
-
-    async def _drain_queue(self, queue: asyncio.Queue):
-        try:
-            while not queue.empty():
-                queue.get_nowait()
-        except asyncio.QueueEmpty:
-            logger.debug(f"Draining {self} queue already empty")
-
-    async def _drain_queues(self):
-        """Drain all frames from upstream and downstream queues."""
-        await self._drain_queue(self._up_queue)
-        await self._drain_queue(self._down_queue)
-
-    async def _handle_interruption(self):
-        """Handle interruption by cancelling tasks, draining queues, and restarting."""
-        await self._cancel()
-        await self._drain_queues()
-        await self._create_tasks()
+        # Process frames in each of the sub-pipelines.
+        for p in self._pipelines:
+            await p.queue_frame(frame, direction)
 
     async def _parallel_push_frame(self, frame: Frame, direction: FrameDirection):
         """Push frames while avoiding duplicates using frame ID tracking."""
@@ -300,52 +157,18 @@ class ParallelPipeline(BasePipeline):
             await self.push_frame(frame, direction)
 
     async def _pipeline_sink_push_frame(self, frame: Frame, direction: FrameDirection):
-        if isinstance(frame, StartFrame):
-            # Decrement counter and check if all pipelines have processed the StartFrame
-            start_frame_counter = self._start_frame_counter.get(frame.id, 0)
-            if start_frame_counter > 0:
-                self._start_frame_counter[frame.id] -= 1
-                start_frame_counter = self._start_frame_counter[frame.id]
+        # Parallel pipeline synchronized frames.
+        if isinstance(frame, (StartFrame, EndFrame)):
+            # Decrement counter.
+            frame_counter = self._frame_counter.get(frame.id, 0)
+            if frame_counter > 0:
+                self._frame_counter[frame.id] -= 1
+                frame_counter = self._frame_counter[frame.id]
 
-            # Only push the StartFrame when all pipelines have processed it
-            if start_frame_counter == 0:
-                self._started = True
-                await self._start(frame)
+            # Only push the frame when all pipelines have processed it.
+            if frame_counter == 0:
                 await self._parallel_push_frame(frame, direction)
+                await self.resume_processing_system_frames()
+                await self.resume_processing_frames()
         else:
-            if self._started:
-                await self._parallel_push_frame(frame, direction)
-            else:
-                await self._down_queue.put(frame)
-
-    async def _process_up_queue(self):
-        """Process upstream frames from all parallel branches."""
-        while True:
-            frame = await self._up_queue.get()
-            await self._parallel_push_frame(frame, FrameDirection.UPSTREAM)
-            self._up_queue.task_done()
-
-    async def _process_down_queue(self):
-        """Process downstream frames with EndFrame coordination.
-
-        Coordinates EndFrames to ensure they are only pushed upstream once
-        all parallel branches have completed processing them.
-        """
-        running = True
-        while running:
-            frame = await self._down_queue.get()
-
-            endframe_counter = self._endframe_counter.get(frame.id, 0)
-
-            # If we have a counter, decrement it.
-            if endframe_counter > 0:
-                self._endframe_counter[frame.id] -= 1
-                endframe_counter = self._endframe_counter[frame.id]
-
-            # If we don't have a counter or we reached 0, push the frame.
-            if endframe_counter == 0:
-                await self._parallel_push_frame(frame, FrameDirection.DOWNSTREAM)
-
-            running = not (endframe_counter == 0 and isinstance(frame, EndFrame))
-
-            self._down_queue.task_done()
+            await self._parallel_push_frame(frame, direction)
diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py
index 5d9dd91c0..48cd4edcf 100644
--- a/src/pipecat/pipeline/pipeline.py
+++ b/src/pipecat/pipeline/pipeline.py
@@ -11,7 +11,7 @@ in sequence and manages frame flow between them, along with helper classes
 for pipeline source and sink operations.
 """
 
-from typing import Callable, Coroutine, List
+from typing import Callable, Coroutine, List, Optional
 
 from pipecat.frames.frames import Frame
 from pipecat.pipeline.base_pipeline import BasePipeline
@@ -26,13 +26,14 @@ class PipelineSource(FrameProcessor):
     provided upstream handler function.
     """
 
-    def __init__(self, upstream_push_frame: Callable[[Frame, FrameDirection], Coroutine]):
+    def __init__(self, upstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs):
         """Initialize the pipeline source.
 
         Args:
             upstream_push_frame: Coroutine function to handle upstream frames.
+            **kwargs: Additional arguments passed to parent class.
         """
-        super().__init__(enable_direct_mode=True)
+        super().__init__(enable_direct_mode=True, **kwargs)
         self._upstream_push_frame = upstream_push_frame
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -59,13 +60,16 @@ class PipelineSink(FrameProcessor):
     provided downstream handler function.
     """
 
-    def __init__(self, downstream_push_frame: Callable[[Frame, FrameDirection], Coroutine]):
+    def __init__(
+        self, downstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs
+    ):
         """Initialize the pipeline sink.
 
         Args:
             downstream_push_frame: Coroutine function to handle downstream frames.
+            **kwargs: Additional arguments passed to parent class.
         """
-        super().__init__(enable_direct_mode=True)
+        super().__init__(enable_direct_mode=True, **kwargs)
         self._downstream_push_frame = downstream_push_frame
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -92,26 +96,60 @@ class Pipeline(BasePipeline):
     provides metrics collection from contained processors.
     """
 
-    def __init__(self, processors: List[FrameProcessor]):
+    def __init__(
+        self,
+        processors: List[FrameProcessor],
+        *,
+        source: Optional[FrameProcessor] = None,
+        sink: Optional[FrameProcessor] = None,
+    ):
         """Initialize the pipeline with a list of processors.
 
         Args:
             processors: List of frame processors to connect in sequence.
+            source: An optional pipeline source processor.
+            sink: An optional pipeline sink processor.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
 
         # Add a source and a sink queue so we can forward frames upstream and
         # downstream outside of the pipeline.
-        self._source = PipelineSource(self.push_frame)
-        self._sink = PipelineSink(self.push_frame)
+        self._source = source or PipelineSource(self.push_frame, name=f"{self}::Source")
+        self._sink = sink or PipelineSink(self.push_frame, name=f"{self}::Sink")
         self._processors: List[FrameProcessor] = [self._source] + processors + [self._sink]
 
         self._link_processors()
 
     #
-    # BasePipeline
+    # Frame processor
     #
 
+    @property
+    def processors(self):
+        """Return the list of sub-processors contained within this processor.
+
+        Only compound processors (e.g. pipelines and parallel pipelines) have
+        sub-processors. Non-compound processors will return an empty list.
+
+        Returns:
+            The list of sub-processors if this is a compound processor.
+        """
+        return self._processors
+
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return [self._source]
+
     def processors_with_metrics(self):
         """Return processors that can generate metrics.
 
@@ -122,17 +160,12 @@ class Pipeline(BasePipeline):
             List of frame processors that can generate metrics.
         """
         services = []
-        for p in self._processors:
-            if isinstance(p, BasePipeline):
-                services.extend(p.processors_with_metrics())
-            elif p.can_generate_metrics():
+        for p in self.processors:
+            if p.can_generate_metrics():
                 services.append(p)
+            services.extend(p.processors_with_metrics())
         return services
 
-    #
-    # Frame processor
-    #
-
     async def setup(self, setup: FrameProcessorSetup):
         """Set up the pipeline and all contained processors.
 
@@ -175,7 +208,5 @@ class Pipeline(BasePipeline):
         """Link all processors in sequence and set their parent."""
         prev = self._processors[0]
         for curr in self._processors[1:]:
-            prev.set_parent(self)
             prev.link(curr)
             prev = curr
-        prev.set_parent(self)
diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py
index 366da824a..b09207a53 100644
--- a/src/pipecat/pipeline/sync_parallel_pipeline.py
+++ b/src/pipecat/pipeline/sync_parallel_pipeline.py
@@ -134,9 +134,35 @@ class SyncParallelPipeline(BasePipeline):
         self._pipelines = []
 
     #
-    # BasePipeline
+    # Frame processor
     #
 
+    @property
+    def processors(self):
+        """Return the list of sub-processors contained within this processor.
+
+        Only compound processors (e.g. pipelines and parallel pipelines) have
+        sub-processors. Non-compound processors will return an empty list.
+
+        Returns:
+            The list of sub-processors if this is a compound processor.
+        """
+        return self._pipelines
+
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return self._sources
+
     def processors_with_metrics(self) -> List[FrameProcessor]:
         """Collect processors that can generate metrics from all parallel pipelines.
 
@@ -145,10 +171,6 @@ class SyncParallelPipeline(BasePipeline):
         """
         return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines))
 
-    #
-    # Frame processor
-    #
-
     async def setup(self, setup: FrameProcessorSetup):
         """Set up the parallel pipeline and all contained processors.
 
@@ -171,29 +193,23 @@ class SyncParallelPipeline(BasePipeline):
             source = SyncParallelPipelineSource(up_queue)
             sink = SyncParallelPipelineSink(down_queue)
 
-            # Create pipeline
-            pipeline = Pipeline(processors)
-            source.link(pipeline)
-            pipeline.link(sink)
-            self._pipelines.append(pipeline)
-
             # Keep track of sources and sinks. We also keep the output queue of
             # the source and the sinks so we can use it later.
             self._sources.append({"processor": source, "queue": down_queue})
             self._sinks.append({"processor": sink, "queue": up_queue})
 
+            # Create pipeline
+            pipeline = Pipeline(processors, source=source, sink=sink)
+            self._pipelines.append(pipeline)
+
         logger.debug(f"Finished creating {self} pipelines")
 
-        await asyncio.gather(*[s["processor"].setup(setup) for s in self._sources])
         await asyncio.gather(*[p.setup(setup) for p in self._pipelines])
-        await asyncio.gather(*[s["processor"].setup(setup) for s in self._sinks])
 
     async def cleanup(self):
         """Clean up the parallel pipeline and all contained processors."""
         await super().cleanup()
-        await asyncio.gather(*[s["processor"].cleanup() for s in self._sources])
         await asyncio.gather(*[p.cleanup() for p in self._pipelines])
-        await asyncio.gather(*[s["processor"].cleanup() for s in self._sinks])
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames through all parallel pipelines with synchronization.
diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 6bc0c3d1e..7ff924531 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -41,8 +41,8 @@ from pipecat.frames.frames import (
 from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
 from pipecat.observers.base_observer import BaseObserver
 from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
-from pipecat.pipeline.base_pipeline import BasePipeline
 from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams
+from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource
 from pipecat.pipeline.task_observer import TaskObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.utils.asyncio.task_manager import (
@@ -101,70 +101,6 @@ class PipelineParams(BaseModel):
     start_metadata: Dict[str, Any] = Field(default_factory=dict)
 
 
-class PipelineTaskSource(FrameProcessor):
-    """Source processor for pipeline tasks that handles frame routing.
-
-    This is the source processor that is linked at the beginning of the
-    pipeline given to the pipeline task. It allows us to easily push frames
-    downstream to the pipeline and also receive upstream frames coming from the
-    pipeline.
-    """
-
-    def __init__(self, up_queue: asyncio.Queue):
-        """Initialize the pipeline task source.
-
-        Args:
-            up_queue: Queue for upstream frame processing.
-            **kwargs: Additional arguments passed to the parent class.
-        """
-        super().__init__(enable_direct_mode=True)
-        self._up_queue = up_queue
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames and route them based on direction.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame flow.
-        """
-        await super().process_frame(frame, direction)
-
-        match direction:
-            case FrameDirection.UPSTREAM:
-                await self._up_queue.put(frame)
-            case FrameDirection.DOWNSTREAM:
-                await self.push_frame(frame, direction)
-
-
-class PipelineTaskSink(FrameProcessor):
-    """Sink processor for pipeline tasks that handles final frame processing.
-
-    This is the sink processor that is linked at the end of the pipeline
-    given to the pipeline task. It allows us to receive downstream frames and
-    act on them, for example, waiting to receive an EndFrame.
-    """
-
-    def __init__(self, down_queue: asyncio.Queue):
-        """Initialize the pipeline task sink.
-
-        Args:
-            down_queue: Queue for downstream frame processing.
-            **kwargs: Additional arguments passed to the parent class.
-        """
-        super().__init__(enable_direct_mode=True)
-        self._down_queue = down_queue
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames and route them to the downstream queue.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame flow.
-        """
-        await super().process_frame(frame, direction)
-        await self._down_queue.put(frame)
-
-
 class PipelineTask(BasePipelineTask):
     """Manages the execution of a pipeline, handling frame processing and task lifecycle.
 
@@ -196,7 +132,7 @@ class PipelineTask(BasePipelineTask):
 
     def __init__(
         self,
-        pipeline: BasePipeline,
+        pipeline: FrameProcessor,
         *,
         params: Optional[PipelineParams] = None,
         additional_span_attributes: Optional[dict] = None,
@@ -244,7 +180,6 @@ class PipelineTask(BasePipelineTask):
                 will be logged if the watchdog timer is not reset before this timeout.
         """
         super().__init__()
-        self._pipeline = pipeline
         self._params = params or PipelineParams()
         self._additional_span_attributes = additional_span_attributes or {}
         self._cancel_on_idle_timeout = cancel_on_idle_timeout
@@ -288,12 +223,6 @@ class PipelineTask(BasePipelineTask):
         # PipelineTask and its frame processors.
         self._task_manager = task_manager or TaskManager()
 
-        # This queue receives frames coming from the pipeline upstream.
-        self._up_queue = WatchdogQueue(self._task_manager)
-        self._process_up_task: Optional[asyncio.Task] = None
-        # This queue receives frames coming from the pipeline downstream.
-        self._down_queue = WatchdogQueue(self._task_manager)
-        self._process_down_task: Optional[asyncio.Task] = None
         # This queue is the queue used to push frames to the pipeline.
         self._push_queue = WatchdogQueue(self._task_manager)
         self._process_push_task: Optional[asyncio.Task] = None
@@ -311,17 +240,13 @@ class PipelineTask(BasePipelineTask):
         # StopFrame) has been received in the down queue.
         self._pipeline_end_event = asyncio.Event()
 
-        # This is a source processor that we connect to the provided
-        # pipeline. This source processor allows up to receive and react to
-        # upstream frames.
-        self._source = PipelineTaskSource(self._up_queue)
-        self._source.link(pipeline)
-
-        # This is a sink processor that we connect to the provided
-        # pipeline. This sink processor allows up to receive and react to
-        # downstream frames.
-        self._sink = PipelineTaskSink(self._down_queue)
-        pipeline.link(self._sink)
+        # This is the final pipeline. It is composed of a source processor,
+        # followed by the user pipeline, and ending with a sink processor. The
+        # source allows us to receive and react to upstream frames, and the sink
+        # allows us to receive and react to downstream frames.
+        source = PipelineSource(self._source_push_frame, name=f"{self}::Source")
+        sink = PipelineSink(self._sink_push_frame, name=f"{self}::Sink")
+        self._pipeline = Pipeline([pipeline], source=source, sink=sink)
 
         # The task observer acts as a proxy to the provided observers. This way,
         # we only need to pass a single observer (using the StartFrame) which
@@ -494,7 +419,7 @@ class PipelineTask(BasePipelineTask):
             # Make sure everything is cleaned up downstream. This is sent
             # out-of-band from the main streaming task which is what we want since
             # we want to cancel right away.
-            await self._source.push_frame(CancelFrame())
+            await self._pipeline.queue_frame(CancelFrame())
             # Wait for CancelFrame to make it throught the pipeline.
             await self._wait_for_pipeline_end()
             # Only cancel the push task, we don't want to be able to process any
@@ -506,12 +431,6 @@ class PipelineTask(BasePipelineTask):
 
     async def _create_tasks(self):
         """Create and start all pipeline processing tasks."""
-        self._process_up_task = self._task_manager.create_task(
-            self._process_up_queue(), f"{self}::_process_up_queue"
-        )
-        self._process_down_task = self._task_manager.create_task(
-            self._process_down_queue(), f"{self}::_process_down_queue"
-        )
         self._process_push_task = self._task_manager.create_task(
             self._process_push_queue(), f"{self}::_process_push_queue"
         )
@@ -545,14 +464,6 @@ class PipelineTask(BasePipelineTask):
             await self._task_manager.cancel_task(self._process_push_task)
             self._process_push_task = None
 
-        if self._process_up_task:
-            await self._task_manager.cancel_task(self._process_up_task)
-            self._process_up_task = None
-
-        if self._process_down_task:
-            await self._task_manager.cancel_task(self._process_down_task)
-            self._process_down_task = None
-
         await self._maybe_cancel_heartbeat_tasks()
         await self._maybe_cancel_idle_task()
 
@@ -606,9 +517,7 @@ class PipelineTask(BasePipelineTask):
             observer=self._observer,
             watchdog_timers_enabled=self._enable_watchdog_timers,
         )
-        await self._source.setup(setup)
         await self._pipeline.setup(setup)
-        await self._sink.setup(setup)
 
     async def _cleanup(self, cleanup_pipeline: bool):
         """Clean up the pipeline task and processors."""
@@ -620,10 +529,8 @@ class PipelineTask(BasePipelineTask):
             self._turn_trace_observer.end_conversation_tracing()
 
         # Cleanup pipeline processors.
-        await self._source.cleanup()
         if cleanup_pipeline:
             await self._pipeline.cleanup()
-        await self._sink.cleanup()
 
     async def _process_push_queue(self):
         """Process frames from the push queue and send them through the pipeline.
@@ -647,16 +554,16 @@ class PipelineTask(BasePipelineTask):
             interruption_strategies=self._params.interruption_strategies,
         )
         start_frame.metadata = self._params.start_metadata
-        await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)
+        await self._pipeline.queue_frame(start_frame)
 
         if self._params.enable_metrics and self._params.send_initial_empty_metrics:
-            await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM)
+            await self._pipeline.queue_frame(self._initial_metrics_frame())
 
         running = True
         cleanup_pipeline = True
         while running:
             frame = await self._push_queue.get()
-            await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
+            await self._pipeline.queue_frame(frame)
             if isinstance(frame, (CancelFrame, EndFrame, StopFrame)):
                 await self._wait_for_pipeline_end()
             running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame))
@@ -664,7 +571,7 @@ class PipelineTask(BasePipelineTask):
             self._push_queue.task_done()
         await self._cleanup(cleanup_pipeline)
 
-    async def _process_up_queue(self):
+    async def _source_push_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames coming upstream from the pipeline.
 
         This is the task that processes frames coming upstream from the
@@ -672,33 +579,29 @@ class PipelineTask(BasePipelineTask):
         pipeline to be stopped (e.g. EndTaskFrame) in which case we would send
         an EndFrame down the pipeline.
         """
-        while True:
-            frame = await self._up_queue.get()
+        if isinstance(frame, self._reached_upstream_types):
+            await self._call_event_handler("on_frame_reached_upstream", frame)
 
-            if isinstance(frame, self._reached_upstream_types):
-                await self._call_event_handler("on_frame_reached_upstream", frame)
-
-            if isinstance(frame, EndTaskFrame):
-                # Tell the task we should end nicely.
-                await self.queue_frame(EndFrame())
-            elif isinstance(frame, CancelTaskFrame):
-                # Tell the task we should end right away.
+        if isinstance(frame, EndTaskFrame):
+            # Tell the task we should end nicely.
+            await self.queue_frame(EndFrame())
+        elif isinstance(frame, CancelTaskFrame):
+            # Tell the task we should end right away.
+            await self.queue_frame(CancelFrame())
+        elif isinstance(frame, StopTaskFrame):
+            # Tell the task we should stop nicely.
+            await self.queue_frame(StopFrame())
+        elif isinstance(frame, ErrorFrame):
+            if frame.fatal:
+                logger.error(f"A fatal error occurred: {frame}")
+                # Cancel all tasks downstream.
                 await self.queue_frame(CancelFrame())
-            elif isinstance(frame, StopTaskFrame):
-                # Tell the task we should stop nicely.
-                await self.queue_frame(StopFrame())
-            elif isinstance(frame, ErrorFrame):
-                if frame.fatal:
-                    logger.error(f"A fatal error occurred: {frame}")
-                    # Cancel all tasks downstream.
-                    await self.queue_frame(CancelFrame())
-                    # Tell the task we should stop.
-                    await self.queue_frame(StopTaskFrame())
-                else:
-                    logger.warning(f"Something went wrong: {frame}")
-            self._up_queue.task_done()
+                # Tell the task we should stop.
+                await self.queue_frame(StopTaskFrame())
+            else:
+                logger.warning(f"Something went wrong: {frame}")
 
-    async def _process_down_queue(self):
+    async def _sink_push_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames coming downstream from the pipeline.
 
         This tasks process frames coming downstream from the pipeline. For
@@ -706,34 +609,30 @@ class PipelineTask(BasePipelineTask):
         processors have handled the EndFrame and therefore we can exit the task
         cleanly.
         """
-        while True:
-            frame = await self._down_queue.get()
+        # Queue received frame to the idle queue so we can monitor idle
+        # pipelines.
+        await self._idle_queue.put(frame)
 
-            # Queue received frame to the idle queue so we can monitor idle
-            # pipelines.
-            await self._idle_queue.put(frame)
+        if isinstance(frame, self._reached_downstream_types):
+            await self._call_event_handler("on_frame_reached_downstream", frame)
 
-            if isinstance(frame, self._reached_downstream_types):
-                await self._call_event_handler("on_frame_reached_downstream", frame)
+        if isinstance(frame, StartFrame):
+            await self._call_event_handler("on_pipeline_started", frame)
 
-            if isinstance(frame, StartFrame):
-                await self._call_event_handler("on_pipeline_started", frame)
-
-                # Start heartbeat tasks now that StartFrame has been processed
-                # by all processors in the pipeline
-                self._maybe_start_heartbeat_tasks()
-            elif isinstance(frame, EndFrame):
-                await self._call_event_handler("on_pipeline_ended", frame)
-                self._pipeline_end_event.set()
-            elif isinstance(frame, StopFrame):
-                await self._call_event_handler("on_pipeline_stopped", frame)
-                self._pipeline_end_event.set()
-            elif isinstance(frame, CancelFrame):
-                await self._call_event_handler("on_pipeline_cancelled", frame)
-                self._pipeline_end_event.set()
-            elif isinstance(frame, HeartbeatFrame):
-                await self._heartbeat_queue.put(frame)
-            self._down_queue.task_done()
+            # Start heartbeat tasks now that StartFrame has been processed
+            # by all processors in the pipeline
+            self._maybe_start_heartbeat_tasks()
+        elif isinstance(frame, EndFrame):
+            await self._call_event_handler("on_pipeline_ended", frame)
+            self._pipeline_end_event.set()
+        elif isinstance(frame, StopFrame):
+            await self._call_event_handler("on_pipeline_stopped", frame)
+            self._pipeline_end_event.set()
+        elif isinstance(frame, CancelFrame):
+            await self._call_event_handler("on_pipeline_cancelled", frame)
+            self._pipeline_end_event.set()
+        elif isinstance(frame, HeartbeatFrame):
+            await self._heartbeat_queue.put(frame)
 
     async def _heartbeat_push_handler(self):
         """Push heartbeat frames at regular intervals."""
@@ -741,7 +640,7 @@ class PipelineTask(BasePipelineTask):
             # Don't use `queue_frame()` because if an EndFrame is queued the
             # task will just stop waiting for the pipeline to finish not
             # allowing more frames to be pushed.
-            await self._source.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time()))
+            await self._pipeline.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time()))
             await asyncio.sleep(self._params.heartbeats_period_secs)
 
     async def _heartbeat_monitor_handler(self):
diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py
index 532ef7977..6e4e44742 100644
--- a/src/pipecat/pipeline/task_observer.py
+++ b/src/pipecat/pipeline/task_observer.py
@@ -13,11 +13,11 @@ the main pipeline execution.
 
 import asyncio
 import inspect
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from attr import dataclass
 
-from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 
@@ -120,14 +120,21 @@ class TaskObserver(BaseObserver):
         for proxy in self._proxies.values():
             await self._task_manager.cancel_task(proxy.task)
 
+    async def on_process_frame(self, data: FramePushed):
+        """Queue frame data for all managed observers.
+
+        Args:
+            data: The frame push event data to distribute to observers.
+        """
+        await self._send_to_proxy(data)
+
     async def on_push_frame(self, data: FramePushed):
         """Queue frame data for all managed observers.
 
         Args:
             data: The frame push event data to distribute to observers.
         """
-        for proxy in self._proxies.values():
-            await proxy.queue.put(data)
+        await self._send_to_proxy(data)
 
     def _started(self) -> bool:
         """Check if the task observer has been started."""
@@ -151,6 +158,10 @@ class TaskObserver(BaseObserver):
             proxies[observer] = proxy
         return proxies
 
+    async def _send_to_proxy(self, data: Any):
+        for proxy in self._proxies.values():
+            await proxy.queue.put(data)
+
     async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver):
         """Handle frame processing for a single observer."""
         on_push_frame_deprecated = False
@@ -169,11 +180,15 @@ class TaskObserver(BaseObserver):
 
         while True:
             data = await queue.get()
-            if on_push_frame_deprecated:
-                await observer.on_push_frame(
-                    data.src, data.dst, data.frame, data.direction, data.timestamp
-                )
-            else:
-                await observer.on_push_frame(data)
+
+            if isinstance(data, FramePushed):
+                if on_push_frame_deprecated:
+                    await observer.on_push_frame(
+                        data.src, data.dst, data.frame, data.direction, data.timestamp
+                    )
+                else:
+                    await observer.on_push_frame(data)
+            elif isinstance(data, FrameProcessed):
+                await observer.on_process_frame(data)
 
             queue.task_done()
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index fbde70840..95b300fce 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -34,14 +34,11 @@ from pipecat.frames.frames import (
     SystemFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
-from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
 from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.asyncio.watchdog_event import WatchdogEvent
-from pipecat.utils.asyncio.watchdog_priority_queue import (
-    WatchdogPriorityCancelSentinel,
-    WatchdogPriorityQueue,
-)
+from pipecat.utils.asyncio.watchdog_priority_queue import WatchdogPriorityQueue
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 from pipecat.utils.base_object import BaseObject
 
@@ -168,8 +165,7 @@ class FrameProcessor(BaseObject):
             watchdog_timeout_secs: Timeout in seconds for watchdog operations.
             **kwargs: Additional arguments passed to parent class.
         """
-        super().__init__(name=name)
-        self._parent: Optional["FrameProcessor"] = None
+        super().__init__(name=name, **kwargs)
         self._prev: Optional["FrameProcessor"] = None
         self._next: Optional["FrameProcessor"] = None
 
@@ -226,6 +222,8 @@ class FrameProcessor(BaseObject):
 
         # The input task that handles all types of frames. It processes system
         # frames right away and queues non-system frames for later processing.
+        self.__should_block_system_frames = False
+        self.__input_event: Optional[asyncio.Event] = None
         self.__input_frame_task: Optional[asyncio.Task] = None
 
         # The process task processes non-system frames.  Non-system frames will
@@ -234,6 +232,7 @@ class FrameProcessor(BaseObject):
         # called. To resume processing frames we need to call
         # `resume_processing_frames()` which will wake up the event.
         self.__should_block_frames = False
+        self.__process_event: Optional[asyncio.Event] = None
         self.__process_frame_task: Optional[asyncio.Task] = None
 
     @property
@@ -254,6 +253,50 @@ class FrameProcessor(BaseObject):
         """
         return self._name
 
+    @property
+    def processors(self) -> List["FrameProcessor"]:
+        """Return the list of sub-processors contained within this processor.
+
+        Only compound processors (e.g. pipelines and parallel pipelines) have
+        sub-processors. Non-compound processors will return an empty list.
+
+        Returns:
+            The list of sub-processors if this is a compound processor.
+        """
+        return []
+
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return []
+
+    @property
+    def next(self) -> Optional["FrameProcessor"]:
+        """Get the next processor.
+
+        Returns:
+            The next processor, or None if there's no next processor.
+        """
+        return self._next
+
+    @property
+    def previous(self) -> Optional["FrameProcessor"]:
+        """Get the previous processor.
+
+        Returns:
+            The previous processor, or None if there's no previous processor.
+        """
+        return self._prev
+
     @property
     def interruptions_allowed(self):
         """Check if interruptions are allowed for this processor.
@@ -313,6 +356,17 @@ class FrameProcessor(BaseObject):
             raise Exception(f"{self} TaskManager is still not initialized.")
         return self._task_manager
 
+    def processors_with_metrics(self):
+        """Return processors that can generate metrics.
+
+        Recursively collects all processors that support metrics generation,
+        including those from nested processors.
+
+        Returns:
+            List of frame processors that can generate metrics.
+        """
+        return []
+
     def can_generate_metrics(self) -> bool:
         """Check if this processor can generate metrics.
 
@@ -482,30 +536,6 @@ class FrameProcessor(BaseObject):
         processor._prev = self
         logger.debug(f"Linking {self} -> {self._next}")
 
-    def get_event_loop(self) -> asyncio.AbstractEventLoop:
-        """Get the event loop used by this processor.
-
-        Returns:
-            The asyncio event loop.
-        """
-        return self.task_manager.get_event_loop()
-
-    def set_parent(self, parent: "FrameProcessor"):
-        """Set the parent processor for this processor.
-
-        Args:
-            parent: The parent processor.
-        """
-        self._parent = parent
-
-    def get_parent(self) -> Optional["FrameProcessor"]:
-        """Get the parent processor.
-
-        Returns:
-            The parent processor, or None if no parent is set.
-        """
-        return self._parent
-
     def get_clock(self) -> BaseClock:
         """Get the clock used by this processor.
 
@@ -519,6 +549,14 @@ class FrameProcessor(BaseObject):
             raise Exception(f"{self} Clock is still not initialized.")
         return self._clock
 
+    def get_event_loop(self) -> asyncio.AbstractEventLoop:
+        """Get the event loop used by this processor.
+
+        Returns:
+            The asyncio event loop.
+        """
+        return self.task_manager.get_event_loop()
+
     async def queue_frame(
         self,
         frame: Frame,
@@ -546,12 +584,23 @@ class FrameProcessor(BaseObject):
         logger.trace(f"{self}: pausing frame processing")
         self.__should_block_frames = True
 
+    async def pause_processing_system_frames(self):
+        """Pause processing of queued system frames."""
+        logger.trace(f"{self}: pausing system frame processing")
+        self.__should_block_system_frames = True
+
     async def resume_processing_frames(self):
         """Resume processing of queued frames."""
         logger.trace(f"{self}: resuming frame processing")
         if self.__process_event:
             self.__process_event.set()
 
+    async def resume_processing_system_frames(self):
+        """Resume processing of queued system frames."""
+        logger.trace(f"{self}: resuming system frame processing")
+        if self.__input_event:
+            self.__input_event.set()
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process a frame.
 
@@ -559,6 +608,16 @@ class FrameProcessor(BaseObject):
             frame: The frame to process.
             direction: The direction of frame flow.
         """
+        if self._observer:
+            timestamp = self._clock.get_time() if self._clock else 0
+            data = FrameProcessed(
+                processor=self,
+                frame=frame,
+                direction=direction,
+                timestamp=timestamp,
+            )
+            await self._observer.on_process_frame(data)
+
         if isinstance(frame, StartFrame):
             await self.__start(frame)
         elif isinstance(frame, StartInterruptionFrame):
@@ -715,6 +774,7 @@ class FrameProcessor(BaseObject):
             return
 
         if not self.__input_frame_task:
+            self.__input_event = WatchdogEvent(self.task_manager)
             self.__input_queue = FrameProcessorQueue(self.task_manager)
             self.__input_frame_task = self.create_task(self.__input_frame_task_handler())
 
@@ -764,6 +824,13 @@ class FrameProcessor(BaseObject):
 
         """
         while True:
+            if self.__should_block_system_frames and self.__input_event:
+                logger.trace(f"{self}: system frame processing paused")
+                await self.__input_event.wait()
+                self.__input_event.clear()
+                self.__should_block_system_frames = False
+                logger.trace(f"{self}: system frame processing resumed")
+
             (frame, direction, callback) = await self.__input_queue.get()
 
             if isinstance(frame, SystemFrame):
@@ -780,7 +847,7 @@ class FrameProcessor(BaseObject):
     async def __process_frame_task_handler(self):
         """Handle non-system frames from the process queue."""
         while True:
-            if self.__should_block_frames:
+            if self.__should_block_frames and self.__process_event:
                 logger.trace(f"{self}: frame processing paused")
                 await self.__process_event.wait()
                 self.__process_event.clear()
diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py
index 2c0f65b2d..a763dbc86 100644
--- a/src/pipecat/tests/utils.py
+++ b/src/pipecat/tests/utils.py
@@ -36,7 +36,7 @@ class SleepFrame(SystemFrame):
         sleep: Duration to sleep in seconds before processing the next frame.
     """
 
-    sleep: float = 0.1
+    sleep: float = 0.2
 
 
 class HeartbeatsObserver(BaseObserver):
@@ -100,7 +100,7 @@ class QueuedFrameProcessor(FrameProcessor):
             queue_direction: The direction of frames to capture (UPSTREAM or DOWNSTREAM).
             ignore_start: Whether to ignore StartFrames when capturing.
         """
-        super().__init__()
+        super().__init__(enable_direct_mode=True)
         self._queue = queue
         self._queue_direction = queue_direction
         self._ignore_start = ignore_start
diff --git a/tests/test_context_aggregators.py b/tests/test_context_aggregators.py
index 232fa59cc..e04487f69 100644
--- a/tests/test_context_aggregators.py
+++ b/tests/test_context_aggregators.py
@@ -102,8 +102,8 @@ class BaseTestUserContextAggregator:
         ]
         expected_down_frames = [
             UserStartedSpeakingFrame,
-            UserStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
+            UserStoppedSpeakingFrame,
         ]
         await run_test(
             aggregator,
@@ -127,8 +127,8 @@ class BaseTestUserContextAggregator:
         ]
         expected_down_frames = [
             UserStartedSpeakingFrame,
-            UserStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
+            UserStoppedSpeakingFrame,
         ]
         await run_test(
             aggregator,
@@ -158,8 +158,8 @@ class BaseTestUserContextAggregator:
             UserStartedSpeakingFrame,
             UserStoppedSpeakingFrame,
             UserStartedSpeakingFrame,
-            UserStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
+            UserStoppedSpeakingFrame,
         ]
         await run_test(
             aggregator,
@@ -298,8 +298,8 @@ class BaseTestUserContextAggregator:
         expected_down_frames = [
             SpeechControlParamsFrame,
             UserStartedSpeakingFrame,
-            UserStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
+            UserStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
         ]
         await run_test(
diff --git a/tests/test_dtmf_aggregator.py b/tests/test_dtmf_aggregator.py
index 40d3ece13..697d50d27 100644
--- a/tests/test_dtmf_aggregator.py
+++ b/tests/test_dtmf_aggregator.py
@@ -53,9 +53,9 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         frames_to_send = [
             InputDTMFFrame(button=KeypadEntry.ONE),
             InputDTMFFrame(button=KeypadEntry.TWO),
-            SleepFrame(sleep=0.2),  # This should trigger timeout
+            SleepFrame(),  # This should trigger timeout
             InputDTMFFrame(button=KeypadEntry.THREE),
-            SleepFrame(sleep=0.2),  # This should trigger another timeout
+            SleepFrame(),  # This should trigger another timeout
         ]
         expected_down_frames = [
             InputDTMFFrame,
@@ -86,10 +86,10 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
             InputDTMFFrame(button=KeypadEntry.ONE),
             InputDTMFFrame(button=KeypadEntry.TWO),
             InputDTMFFrame(button=KeypadEntry.POUND),  # First sequence
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             InputDTMFFrame(button=KeypadEntry.FOUR),
             InputDTMFFrame(button=KeypadEntry.FIVE),
-            SleepFrame(sleep=0.3),  # Second sequence via timeout
+            SleepFrame(),  # Second sequence via timeout
         ]
         expected_down_frames = [
             InputDTMFFrame,
@@ -120,7 +120,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
         frames_to_send = [
             InputDTMFFrame(button=KeypadEntry.ONE),
             InputDTMFFrame(button=KeypadEntry.TWO),
-            SleepFrame(sleep=0.1),  # Allow time for aggregation
+            SleepFrame(),  # Allow time for aggregation
             EndFrame(),
         ]
         expected_down_frames = [
diff --git a/tests/test_stt_mute_filter.py b/tests/test_stt_mute_filter.py
index 161962224..3d707dca7 100644
--- a/tests/test_stt_mute_filter.py
+++ b/tests/test_stt_mute_filter.py
@@ -124,14 +124,14 @@ class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
         frames_to_send = [
             # Bot speaking - should mute
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),  # Wait for StartedSpeaking to process
+            SleepFrame(),  # Wait for StartedSpeaking to process
             InterimTranscriptionFrame(
                 user_id="user1", text="This should be suppressed", timestamp="1234567890"
             ),
             TranscriptionFrame(
                 user_id="user1", text="This should be suppressed", timestamp="1234567890"
             ),
-            SleepFrame(sleep=0.1),  # Wait for transcription frames to queue
+            SleepFrame(),  # Wait for transcription frames to queue
             BotStoppedSpeakingFrame(),
             # Bot not speaking - should pass through
             InterimTranscriptionFrame(
diff --git a/tests/test_transcript_processor.py b/tests/test_transcript_processor.py
index db663e086..a2a057a13 100644
--- a/tests/test_transcript_processor.py
+++ b/tests/test_transcript_processor.py
@@ -129,13 +129,13 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
         # Create test frames simulating bot speaking multiple text chunks
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),  # Wait for StartedSpeaking to process
+            SleepFrame(),  # Wait for StartedSpeaking to process
             TTSTextFrame(text="Hello"),
             TTSTextFrame(text="world!"),
             TTSTextFrame(text="How"),
             TTSTextFrame(text="are"),
             TTSTextFrame(text="you?"),
-            SleepFrame(sleep=0.1),  # Wait for text frames to queue
+            SleepFrame(),  # Wait for text frames to queue
             BotStoppedSpeakingFrame(),
         ]
 
@@ -151,8 +151,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
             TTSTextFrame,
             TTSTextFrame,
             TTSTextFrame,
-            BotStoppedSpeakingFrame,
             TranscriptionUpdateFrame,
+            BotStoppedSpeakingFrame,
         ]
 
         # Run test
@@ -179,7 +179,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
         self.assertIsNotNone(message.timestamp)
 
         # All frames should be passed through in order, with update at end
-        downstream_update = cast(TranscriptionUpdateFrame, received_frames[-1])
+        downstream_update = cast(TranscriptionUpdateFrame, received_frames[-2])
         self.assertEqual(downstream_update.messages[0].content, "Hello world! How are you?")
 
     async def test_empty_text_handling(self):
@@ -194,7 +194,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
 
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             TTSTextFrame(text=""),  # Empty text
             TTSTextFrame(text="   "),  # Just whitespace
             TTSTextFrame(text="\n"),  # Just newline
@@ -234,16 +234,16 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
         # Simulate bot being interrupted mid-sentence
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             TTSTextFrame(text="Hello"),
             TTSTextFrame(text="world!"),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             StartInterruptionFrame(),  # User interrupts here
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             BotStartedSpeakingFrame(),
             TTSTextFrame(text="New"),
             TTSTextFrame(text="response"),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             BotStoppedSpeakingFrame(),
         ]
 
@@ -257,8 +257,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
             BotStartedSpeakingFrame,
             TTSTextFrame,  # "New"
             TTSTextFrame,  # "response"
-            BotStoppedSpeakingFrame,
             TranscriptionUpdateFrame,  # Second message
+            BotStoppedSpeakingFrame,
         ]
 
         # Run test
@@ -298,7 +298,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
 
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             TTSTextFrame(text="Hello"),
             TTSTextFrame(text="world"),
             # Pipeline ends here; run_test will automatically send EndFrame
@@ -337,10 +337,10 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
 
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             TTSTextFrame(text="Hello"),
             TTSTextFrame(text="world"),
-            SleepFrame(sleep=0.1),  # Ensure messages are processed
+            SleepFrame(),  # Ensure messages are processed
             CancelFrame(),
         ]
 
@@ -400,7 +400,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
         # Test assistant processor
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             TTSTextFrame(text="Assistant"),
             TTSTextFrame(text="message"),
             BotStoppedSpeakingFrame(),
@@ -440,7 +440,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
         # Test the specific pattern shared
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            SleepFrame(sleep=0.1),
+            SleepFrame(),
             TTSTextFrame(text="Hello"),
             TTSTextFrame(text=" there"),
             TTSTextFrame(text="!"),