diff --git a/api/services/pipecat/pipeline_engine_callbacks_processor.py b/api/services/pipecat/pipeline_engine_callbacks_processor.py index 0445f13..8c04880 100644 --- a/api/services/pipecat/pipeline_engine_callbacks_processor.py +++ b/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): diff --git a/api/services/workflow/pipecat_engine_callbacks.py b/api/services/workflow/pipecat_engine_callbacks.py index 87ff06e..0c80704 100644 --- a/api/services/workflow/pipecat_engine_callbacks.py +++ b/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 diff --git a/api/tests/test_pipecat_engine_callbacks.py b/api/tests/test_pipecat_engine_callbacks.py new file mode 100644 index 0000000..d5f6a36 --- /dev/null +++ b/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, + )