feat: abort immediately on max call duration exceed

api/services/pipecat/pipeline_engine_callbacks_processor.py

@@ -67,7 +67,7 @@ class PipelineEngineCallbacksProcessor(FrameProcessor):
                     self._end_task_frame_pushed = True
                 else:
                     logger.debug(
-                        "Max call duration exceeded. Skipping EndTaskFrame since already sent"
+                        "Max call duration exceeded. Skipping termination since already requested"
                     )
 
     async def _generation_started(self):

api/services/workflow/pipecat_engine_callbacks.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 """Callback factory helpers for :pyclass:`~api.services.workflow.pipecat_engine.PipecatEngine`.
 
 Each helper takes a :class:`PipecatEngine` instance and returns an async
@@ -10,6 +8,8 @@ encapsulating the callback implementations here for easier maintenance and
 unit-testing.
 """
 
+from __future__ import annotations
+
 import re
 from typing import TYPE_CHECKING
 
@@ -73,11 +73,14 @@ def create_user_idle_handler(engine: "PipecatEngine") -> UserIdleHandler:
 
 
 def create_max_duration_callback(engine: "PipecatEngine"):
-    """Return a callback that ends the task when the max call duration is exceeded."""
+    """Return a callback that cancels the task when the hard call limit is exceeded."""
 
     async def handle_max_duration():
         logger.debug("Max call duration exceeded. Terminating call")
-        await engine.end_call_with_reason(EndTaskReason.CALL_DURATION_EXCEEDED.value)
+        await engine.end_call_with_reason(
+            EndTaskReason.CALL_DURATION_EXCEEDED.value,
+            abort_immediately=True,
+        )
 
     return handle_max_duration
 

api/tests/test_pipecat_engine_callbacks.py

@@ -0,0 +1,19 @@
+from unittest.mock import AsyncMock
+
+import pytest
+from pipecat.utils.enums import EndTaskReason
+
+from api.services.workflow.pipecat_engine_callbacks import create_max_duration_callback
+
+
+@pytest.mark.asyncio
+async def test_max_duration_callback_aborts_immediately():
+    engine = AsyncMock()
+
+    callback = create_max_duration_callback(engine)
+    await callback()
+
+    engine.end_call_with_reason.assert_awaited_once_with(
+        EndTaskReason.CALL_DURATION_EXCEEDED.value,
+        abort_immediately=True,
+    )