From de8ba685893ea2b09638936d09f7a6f5f130e7c3 Mon Sep 17 00:00:00 2001 From: OmercohenAviv Date: Fri, 27 Mar 2026 19:41:06 +0300 Subject: [PATCH 1/5] Fix heartbeat monitor timeout not respecting custom heartbeat interval The heartbeat monitor timeout (`HEARTBEAT_MONITOR_SECS`) was a static module-level constant that never derived from the user-configurable `heartbeats_period_secs`. This meant overriding the heartbeat interval had no effect on the monitor window, causing spurious warnings or delayed detection depending on the configured interval. Add a new `heartbeats_monitor_secs` parameter to `PipelineParams` so the monitor timeout is independently configurable (defaults to 10s). The monitor handler now reads from the instance param instead of the hard-coded constant. Made-with: Cursor --- src/pipecat/pipeline/task.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 90976b52c..39b4472a6 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -52,7 +52,7 @@ from pipecat.utils.tracing.setup import is_tracing_available 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 @@ -111,6 +111,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: Strategies for bot interruption behavior. observers: [deprecated] Use `observers` arg in `PipelineTask` class. @@ -131,6 +133,7 @@ class PipelineParams(BaseModel): enable_metrics: bool = False enable_usage_metrics: bool = False heartbeats_period_secs: float = HEARTBEAT_SECS + heartbeats_monitor_secs: Optional[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 @@ -804,9 +807,9 @@ 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 while True: try: + wait_time = self._params.heartbeats_monitor_secs frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) process_time = (self._clock.get_time() - frame.timestamp) / 1_000_000_000 logger.trace(f"{self}: heartbeat frame processed in {process_time} seconds") From dccd98ec8afac337dd76bbe5e28251ace7de4859 Mon Sep 17 00:00:00 2001 From: OmercohenAviv Date: Sat, 28 Mar 2026 11:53:51 +0300 Subject: [PATCH 2/5] test --- tests/test_pipeline.py | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 3c7f50453..579b77b03 100644 --- a/tests/test_pipeline.py +++ b/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, @@ -298,6 +301,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]) From 33fb8852e65751ae5f96a915e59853dab981d6f8 Mon Sep 17 00:00:00 2001 From: OmercohenAviv Date: Sat, 28 Mar 2026 12:05:30 +0300 Subject: [PATCH 3/5] ruff --- src/pipecat/pipeline/task.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 41427ef4d..df9d5dbe9 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -149,7 +149,7 @@ class PipelineParams(BaseModel): enable_metrics: bool = False enable_usage_metrics: bool = False heartbeats_period_secs: float = HEARTBEAT_SECS - heartbeats_monitor_secs: Optional[float] = HEARTBEAT_MONITOR_SECS + heartbeats_monitor_secs: Optional[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 From 2724ef6d6fdeca8692a407e18d167997c7eba49e Mon Sep 17 00:00:00 2001 From: OmercohenAviv Date: Sat, 28 Mar 2026 12:12:02 +0300 Subject: [PATCH 4/5] non optional --- src/pipecat/pipeline/task.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index df9d5dbe9..5f19cf86f 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -149,7 +149,7 @@ class PipelineParams(BaseModel): enable_metrics: bool = False enable_usage_metrics: bool = False heartbeats_period_secs: float = HEARTBEAT_SECS - heartbeats_monitor_secs: Optional[float] = HEARTBEAT_MONITOR_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 From f2e0f5d20c3e5d979e0a9d2c57bbf562ad94b3b1 Mon Sep 17 00:00:00 2001 From: OmercohenAviv Date: Sun, 29 Mar 2026 00:05:21 +0300 Subject: [PATCH 5/5] move wait_time out of loop --- src/pipecat/pipeline/task.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 5f19cf86f..4b471c08e 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -967,9 +967,9 @@ 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 = self._params.heartbeats_monitor_secs while True: try: - wait_time = self._params.heartbeats_monitor_secs frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) process_time = (self._clock.get_time() - frame.timestamp) / 1_000_000_000 logger.trace(f"{self}: heartbeat frame processed in {process_time} seconds")