fix: fix tests

api/services/pipecat/run_pipeline.py

@@ -51,6 +51,7 @@ from api.services.pipecat.tracing_config import (
     ensure_tracing,
 )
 from api.services.pipecat.transport_setup import create_webrtc_transport
+from api.services.pipecat.worker_runner import run_pipeline_worker
 from api.services.pipecat.ws_sender_registry import get_ws_sender
 from api.services.telephony import registry as telephony_registry
 from api.services.workflow.dto import ReactFlowDTO
@@ -87,7 +88,6 @@ 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()
@@ -821,9 +821,7 @@ async def _run_pipeline(
 
     try:
         # Run the pipeline
-        loop = asyncio.get_running_loop()
-        params = WorkerParams(loop=loop)
-        await task.run(params)
+        await run_pipeline_worker(task)
         logger.info(f"Task completed for run {workflow_run_id}")
     except asyncio.CancelledError:
         logger.warning("Received CancelledError in _run_pipeline")

api/services/pipecat/worker_runner.py

@@ -0,0 +1,36 @@
+import asyncio
+
+from pipecat.pipeline.worker import PipelineWorker
+from pipecat.workers.runner import WorkerRunner
+
+
+async def run_pipeline_worker(
+    worker: PipelineWorker,
+    *,
+    handle_sigint: bool = False,
+    handle_sigterm: bool = False,
+    auto_end: bool = True,
+) -> None:
+    """Run a pipeline worker through the v1.3 worker runner lifecycle."""
+    runner = WorkerRunner(handle_sigint=handle_sigint, handle_sigterm=handle_sigterm)
+    await runner.add_workers(worker)
+    await runner.run(auto_end=auto_end)
+
+
+async def wait_for_pipeline_worker_started(
+    worker: PipelineWorker,
+    *,
+    timeout: float = 3.0,
+    run_task: asyncio.Task | None = None,
+) -> None:
+    """Wait until a pipeline worker has fired its stable start lifecycle."""
+
+    async def _wait_until_started():
+        while worker.started_at is None:
+            if run_task and run_task.done():
+                await run_task
+            if worker.has_finished():
+                raise RuntimeError("PipelineWorker finished before starting")
+            await asyncio.sleep(0.01)
+
+    await asyncio.wait_for(_wait_until_started(), timeout=timeout)

api/services/workflow/text_chat_runner.py

@@ -22,7 +22,6 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMAssistantAggregatorParams,
@@ -45,6 +44,10 @@ from api.services.pipecat.tracing_config import (
     build_remote_parent_context,
     get_trace_url,
 )
+from api.services.pipecat.worker_runner import (
+    run_pipeline_worker,
+    wait_for_pipeline_worker_started,
+)
 from api.services.workflow.dto import ReactFlowDTO
 from api.services.workflow.pipecat_engine import PipecatEngine
 from api.services.workflow.workflow_graph import WorkflowGraph
@@ -534,8 +537,7 @@ async def execute_text_chat_pending_turn(
         conversation_type="text",
         additional_span_attributes=trace_span_attributes,
     )
-    runner = PipelineRunner(handle_sigint=False, handle_sigterm=False)
-    runner_task = asyncio.create_task(runner.run(task))
+    runner_task = asyncio.create_task(run_pipeline_worker(task))
 
     engine.set_task(task)
     engine.set_audio_config(audio_config)
@@ -548,7 +550,7 @@ async def execute_text_chat_pending_turn(
     )
 
     try:
-        await asyncio.wait_for(task._pipeline_start_event.wait(), timeout=5.0)
+        await wait_for_pipeline_worker_started(task, timeout=5.0, run_task=runner_task)
 
         await engine.initialize()