Merge pull request #4187 from OmerCohenAviv/fix/heartbeat-monitor-configurable

Fix heartbeat monitor timeout not respecting custom heartbeat interval

src/pipecat/pipeline/task.py

@@ -57,7 +57,7 @@ from pipecat.utils.tracing.tracing_context import TracingContext
 from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
 
 HEARTBEAT_SECS = 1.0
-HEARTBEAT_MONITOR_SECS = HEARTBEAT_SECS * 10
+HEARTBEAT_MONITOR_SECS = 10.0
 
 IDLE_TIMEOUT_SECS = 300
 
@@ -123,6 +123,8 @@ class PipelineParams(BaseModel):
         enable_metrics: Whether to enable metrics collection.
         enable_usage_metrics: Whether to enable usage metrics.
         heartbeats_period_secs: Period between heartbeats in seconds.
+        heartbeats_monitor_secs: Timeout (in seconds) before warning about
+            missed heartbeats. Defaults to 10 seconds.
         interruption_strategies: [deprecated] Strategies for bot interruption behavior.
 
             .. deprecated:: 0.0.99
@@ -147,6 +149,7 @@ class PipelineParams(BaseModel):
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
     heartbeats_period_secs: float = HEARTBEAT_SECS
+    heartbeats_monitor_secs: float = HEARTBEAT_MONITOR_SECS
     interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
     observers: List[BaseObserver] = Field(default_factory=list)
     report_only_initial_ttfb: bool = False
@@ -964,7 +967,7 @@ class PipelineTask(BasePipelineTask):
         the time that a heartbeat frame takes to processes, that is how long it
         takes for the heartbeat frame to traverse all the pipeline.
         """
-        wait_time = HEARTBEAT_MONITOR_SECS
+        wait_time = self._params.heartbeats_monitor_secs
         while True:
             try:
                 frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time)

tests/test_pipeline.py

@@ -5,9 +5,12 @@
 #
 
 import asyncio
+import io
 import time
 import unittest
 
+from loguru import logger
+
 from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
@@ -383,6 +386,45 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
             pass
         assert heartbeats_counter == expected_heartbeats
 
+    async def test_heartbeat_monitor_respects_custom_timeout(self):
+        """Verify the heartbeat monitor uses heartbeats_monitor_secs from params."""
+
+        class HeartbeatBlocker(FrameProcessor):
+            async def process_frame(self, frame: Frame, direction: FrameDirection):
+                await super().process_frame(frame, direction)
+                if not isinstance(frame, HeartbeatFrame):
+                    await self.push_frame(frame, direction)
+
+        log_output = io.StringIO()
+        handler_id = logger.add(log_output, level="WARNING", format="{message}")
+
+        custom_monitor_secs = 0.3
+
+        try:
+            pipeline = Pipeline([HeartbeatBlocker()])
+            task = PipelineTask(
+                pipeline,
+                params=PipelineParams(
+                    enable_heartbeats=True,
+                    heartbeats_period_secs=0.1,
+                    heartbeats_monitor_secs=custom_monitor_secs,
+                ),
+                cancel_on_idle_timeout=False,
+            )
+
+            try:
+                await asyncio.wait_for(
+                    task.run(PipelineTaskParams(loop=asyncio.get_event_loop())),
+                    timeout=0.6,
+                )
+            except asyncio.TimeoutError:
+                pass
+
+            log_text = log_output.getvalue()
+            assert f"more than {custom_monitor_secs} seconds" in log_text
+        finally:
+            logger.remove(handler_id)
+
     async def test_idle_task(self):
         identity = IdentityFilter()
         pipeline = Pipeline([identity])