From de8ba685893ea2b09638936d09f7a6f5f130e7c3 Mon Sep 17 00:00:00 2001 From: OmercohenAviv Date: Fri, 27 Mar 2026 19:41:06 +0300 Subject: [PATCH] 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")