chore: rename PipelineTask to PipelineWorker

api/services/pipecat/event_handlers.py

@@ -21,7 +21,7 @@ from api.tasks.function_names import FunctionNames
 from pipecat.frames.frames import (
     Frame,
 )
-from pipecat.pipeline.task import PipelineTask
+from pipecat.pipeline.worker import PipelineWorker
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.utils.enums import EndTaskReason
 
@@ -58,7 +58,7 @@ async def _capture_call_event(
 
 
 def register_event_handlers(
-    task: PipelineTask,
+    task: PipelineWorker,
     transport,
     workflow_run_id: int,
     engine: PipecatEngine,
@@ -184,13 +184,13 @@ def register_event_handlers(
         )
 
     @task.event_handler("on_pipeline_started")
-    async def on_pipeline_started(_task: PipelineTask, _frame: Frame):
+    async def on_pipeline_started(_task: PipelineWorker, _frame: Frame):
         logger.debug("In on_pipeline_started callback handler")
         ready_state["pipeline_started"] = True
         await maybe_trigger_initial_response()
 
     @task.event_handler("on_pipeline_error")
-    async def on_pipeline_error(_task: PipelineTask, frame: Frame):
+    async def on_pipeline_error(_task: PipelineWorker, frame: Frame):
         logger.warning(f"Pipeline error for workflow run {workflow_run_id}: {frame}")
         try:
             workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id)
@@ -218,7 +218,7 @@ def register_event_handlers(
 
     @task.event_handler("on_pipeline_finished")
     async def on_pipeline_finished(
-        task: PipelineTask,
+        task: PipelineWorker,
         _frame: Frame,
     ):
         logger.debug(f"In on_pipeline_finished callback handler")

api/services/pipecat/pipeline_builder.py

@@ -4,7 +4,7 @@ from loguru import logger
 
 from api.services.pipecat.audio_config import AudioConfig
 from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.utils.run_context import turn_var
@@ -194,7 +194,7 @@ def create_pipeline_task(
             f"out: {audio_config.transport_out_sample_rate}Hz"
         )
 
-    task = PipelineTask(
+    task = PipelineWorker(
         pipeline,
         params=pipeline_params,
         enable_tracing=True,

api/services/pipecat/run_pipeline.py

@@ -61,7 +61,6 @@ from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnal
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.extensions.voicemail.voicemail_detector import VoicemailDetector
-from pipecat.pipeline.base_task import PipelineTaskParams
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
     LLMContextAggregatorPair,
@@ -88,6 +87,7 @@ from pipecat.turns.user_stop import (
 from pipecat.turns.user_turn_strategies import UserTurnStrategies
 from pipecat.utils.enums import EndTaskReason, RealtimeFeedbackType
 from pipecat.utils.run_context import set_current_org_id, set_current_run_id
+from pipecat.workers.base_worker import WorkerParams
 
 # Setup tracing if enabled
 ensure_tracing()
@@ -822,7 +822,7 @@ async def _run_pipeline(
     try:
         # Run the pipeline
         loop = asyncio.get_running_loop()
-        params = PipelineTaskParams(loop=loop)
+        params = WorkerParams(loop=loop)
         await task.run(params)
         logger.info(f"Task completed for run {workflow_run_id}")
     except asyncio.CancelledError:

api/services/workflow/pipecat_engine.py

@@ -10,7 +10,7 @@ from pipecat.frames.frames import (
     LLMContextFrame,
     TTSSpeakFrame,
 )
-from pipecat.pipeline.task import PipelineTask
+from pipecat.pipeline.worker import PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.services.settings import LLMSettings
@@ -60,7 +60,7 @@ class PipecatEngine:
     def __init__(
         self,
         *,
-        task: Optional[PipelineTask] = None,
+        task: Optional[PipelineWorker] = None,
         llm: Optional["LLMService"] = None,
         inference_llm: Optional["LLMService"] = None,
         context: Optional[LLMContext] = None,
@@ -842,7 +842,7 @@ class PipecatEngine:
         """
         self.context = context
 
-    def set_task(self, task: PipelineTask) -> None:
+    def set_task(self, task: PipelineWorker) -> None:
         """Set the pipeline task.
 
         This allows setting the task after the engine has been created,

api/tests/integrations/_run_pipeline_helpers.py

@@ -15,7 +15,7 @@ Provided here:
 - ``NoopFeedbackObserver``: a ``RealtimeFeedbackObserver`` stand-in with
   no WebSocket / clock-task side effects.
 - ``patch_run_pipeline_externals``: ``contextmanager`` that applies the
-  full patch set and captures the constructed ``PipelineTask`` for the
+  full patch set and captures the constructed ``PipelineWorker`` for the
   caller. Optional ``llm`` / ``tts`` arguments inject preconfigured
   mocks; otherwise blank ``MockLLMService`` / ``MockTTSService``
   instances are constructed per-call.
@@ -84,10 +84,10 @@ def patch_run_pipeline_externals(
     tts: MockTTSService | None = None,
 ):
     """Patch the externally-talking pieces of ``_run_pipeline`` and capture
-    the constructed ``PipelineTask`` so tests can drive it from outside.
+    the constructed ``PipelineWorker`` so tests can drive it from outside.
 
     Args:
-        captured_task: A list the constructed ``PipelineTask`` is appended
+        captured_task: A list the constructed ``PipelineWorker`` is appended
             to. Tests read ``captured_task[0]`` to get a handle on the task
             (to wait on its start event, queue frames, cancel it, etc.).
         llm: Optional pre-built ``MockLLMService``. When given, every call
@@ -168,7 +168,7 @@ def patch_run_pipeline_externals(
                 return_value="completed",
             )
         )
-        # Capture the PipelineTask so the test can drive it from outside.
+        # Capture the PipelineWorker so the test can drive it from outside.
         stack.enter_context(
             patch(
                 "api.services.pipecat.run_pipeline.create_pipeline_task",

api/tests/integrations/test_run_pipeline.py

@@ -2,7 +2,7 @@
 
 Drives the actual ``_run_pipeline`` against the test database with real
 DB rows (organization, user, user configuration, workflow, workflow run)
-and pipecat's real ``MockTransport`` / ``Pipeline`` / ``PipelineTask``.
+and pipecat's real ``MockTransport`` / ``Pipeline`` / ``PipelineWorker``.
 The only patches are for things that talk to genuinely external systems;
 those are applied via ``patch_run_pipeline_externals`` from the shared
 helpers module.

api/tests/integrations/test_run_pipeline_text_greeting.py

@@ -191,7 +191,7 @@ async def _run_test_body(workflow_run_setup, db_session) -> None:
             )
 
             # Locate the assistant aggregator's LLM context (downstream of TTS).
-            # The PipelineTask wraps the user's pipeline inside another Pipeline,
+            # The PipelineWorker wraps the user's pipeline inside another Pipeline,
             # so we walk the tree recursively.
             assistant_aggregator = _find_processor_by_class_name(
                 pipeline_task, "LLMAssistantAggregator"

api/tests/test_pipecat_engine_context_update.py

@@ -21,7 +21,7 @@ import pytest
 from pipecat.frames.frames import LLMContextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -116,7 +116,7 @@ async def run_pipeline_and_capture_context(
     )
 
     # Create pipeline task
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
 
     engine.set_task(task)
 

api/tests/test_pipecat_engine_end_call.py

@@ -26,7 +26,7 @@ import pytest
 from pipecat.frames.frames import Frame, LLMContextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -112,7 +112,7 @@ async def create_engine_with_tracking(
     mock_llm: MockLLMService,
     test_helper: EndCallTestHelper,
     generate_audio: bool = True,
-) -> tuple[PipecatEngine, MockTTSService, MockTransport, PipelineTask]:
+) -> tuple[PipecatEngine, MockTTSService, MockTransport, PipelineWorker]:
     """Create a PipecatEngine with tracking for end call behavior.
 
     Args:
@@ -222,7 +222,7 @@ async def create_engine_with_tracking(
     )
 
     # Create pipeline task
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
 
     engine.set_task(task)
 

api/tests/test_pipecat_engine_node_switch_with_user_speech.py

@@ -25,7 +25,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -119,7 +119,7 @@ async def create_test_pipeline(
     workflow: WorkflowGraph,
     mock_llm: MockLLMService,
     user_speech_initial_delay: float = 0.01,
-) -> tuple[PipecatEngine, MockTransport, PipelineTask]:
+) -> tuple[PipecatEngine, MockTransport, PipelineWorker]:
     """Create a PipecatEngine with full pipeline for testing node switch scenarios.
 
     The pipeline includes a UserSpeechInjector processor that injects
@@ -208,7 +208,7 @@ async def create_test_pipeline(
     )
 
     # Create pipeline task
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
 
     engine.set_task(task)
 

api/tests/test_pipecat_engine_tool_calls.py

@@ -12,7 +12,7 @@ import pytest
 from pipecat.frames.frames import LLMContextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -107,7 +107,7 @@ async def run_pipeline_with_tool_calls(
     )
 
     # Create a real pipeline task
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
 
     engine.set_task(task)
 

api/tests/test_pipecat_engine_transition_mute.py

@@ -16,7 +16,7 @@ import pytest
 from pipecat.frames.frames import LLMContextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -99,7 +99,7 @@ async def _build_engine_and_pipeline(
         ]
     )
 
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
     engine.set_task(task)
 
     return engine, task, function_call_mute_strategy, user_context_aggregator

api/tests/test_pipecat_engine_variable_extraction.py

@@ -19,7 +19,7 @@ import pytest
 from pipecat.frames.frames import LLMContextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -142,7 +142,7 @@ class TestVariableExtractionDuringTransitions:
         )
 
         # Create pipeline task
-        task = PipelineTask(
+        task = PipelineWorker(
             pipeline,
             params=PipelineParams(),
             enable_rtvi=False,

api/tests/test_pipeline_cancellation.py

@@ -8,10 +8,10 @@ from pipecat.frames.frames import (
     InterruptionTaskFrame,
     LLMRunFrame,
 )
-from pipecat.pipeline.base_task import PipelineTaskParams
 from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.task import PipelineTask
+from pipecat.pipeline.worker import PipelineWorker
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.workers.base_worker import WorkerParams
 
 
 class MockTransport(FrameProcessor):
@@ -51,11 +51,11 @@ async def test_interruption_with_blocked_end_frame():
     transport = MockTransport()
     pipeline = Pipeline([transport, busy_wait_processor])
 
-    task = PipelineTask(pipeline, enable_rtvi=False)
+    task = PipelineWorker(pipeline, enable_rtvi=False)
 
     async def run_pipeline():
         loop = asyncio.get_running_loop()
-        params = PipelineTaskParams(loop=loop)
+        params = WorkerParams(loop=loop)
         await task.run(params=params)
 
     async def queue_frame():

api/tests/test_text_and_audio_playback.py

@@ -21,7 +21,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -212,7 +212,7 @@ async def run_pipeline_and_capture_frames(
         engine.set_transport_output(transport_output)
 
     pipeline = Pipeline([llm, tts, transport_output, context_aggregator.assistant()])
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
     engine.set_task(task)
 
     # Spy on task.queue_frame and transport_output.queue_frame to capture

api/tests/test_tts_endframe_with_audio_write_failure.py

@@ -35,7 +35,7 @@ import pytest
 from pipecat.frames.frames import LLMContextFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -62,7 +62,7 @@ async def create_test_pipeline_with_failing_transport(
     workflow: WorkflowGraph,
     mock_llm: MockLLMService,
     fail_after_n_frames: int = 0,
-) -> tuple[PipecatEngine, MockTTSService, MockTransport, PipelineTask]:
+) -> tuple[PipecatEngine, MockTTSService, MockTransport, PipelineWorker]:
     """Create a PipecatEngine with failing output transport for testing.
 
     Uses the real MockTransport which now extends BaseOutputTransport and uses
@@ -152,7 +152,7 @@ async def create_test_pipeline_with_failing_transport(
     )
 
     # Create pipeline task
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
 
     engine.set_task(task)
 

api/tests/test_user_idle_handler.py

@@ -24,7 +24,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -100,7 +100,7 @@ async def create_pipeline_with_speech_injection(
     speeches: list[str],
     user_idle_timeout: float = 0.2,
     mock_audio_duration_ms: int = 400,
-) -> tuple[PipecatEngine, PipelineTask, object]:
+) -> tuple[PipecatEngine, PipelineWorker, object]:
     """Create a pipeline with user speech injection and idle handling.
 
     Sets up a realistic pipeline with:
@@ -194,7 +194,7 @@ async def create_pipeline_with_speech_injection(
         ]
     )
 
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
     engine.set_task(task)
 
     return engine, task, user_idle_handler

api/tests/test_user_muting_during_bot_speech.py

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -125,7 +125,7 @@ async def create_engine_for_mute_test(
     PipecatEngine,
     MockTTSService,
     MockTransport,
-    PipelineTask,
+    PipelineWorker,
     LLMUserAggregator,
     BotSpeakingObserverProcessor,
 ]:
@@ -196,7 +196,7 @@ async def create_engine_for_mute_test(
         ]
     )
 
-    task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+    task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
     engine.set_task(task)
 
     return engine, tts, mock_transport, task, user_context_aggregator, observer

api/tests/test_voicemail_detector.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -161,7 +161,7 @@ class TestVoicemailDetectorWithUserAggregator:
             ]
         )
 
-        task = PipelineTask(pipeline, params=PipelineParams(), enable_rtvi=False)
+        task = PipelineWorker(pipeline, params=PipelineParams(), enable_rtvi=False)
         runner = PipelineRunner()
 
         async def run_pipeline():

pipecat

@@ -1 +1 @@
-Subproject commit a8458879e6c950007b0753a33652f744fca5cdb5
+Subproject commit 1e145e4267df535b64547ed229d6234a99bd8843