chore: drain active calls before rolling updates

api/routes/main.py

@@ -125,3 +125,22 @@ async def health() -> HealthResponse:
             STACK_PUBLISHABLE_CLIENT_KEY if is_stack else None
         ),
     )
+
+
+class ActiveCallsResponse(BaseModel):
+    active_calls: int
+
+
+@router.get("/health/active-calls", response_model=ActiveCallsResponse)
+async def active_calls() -> ActiveCallsResponse:
+    """In-flight call count for THIS worker — the drain signal for deploys.
+
+    A deploy orchestrator polls this per worker and waits for zero before
+    sending SIGTERM, because uvicorn force-closes live call WebSockets (close
+    code 1012) on SIGTERM and would cut calls mid-conversation otherwise. The
+    count is per-process: one uvicorn per VM port (scripts/rolling_update.sh)
+    or per Kubernetes pod (preStop hook). See api/services/pipecat/active_calls.py.
+    """
+    from api.services.pipecat.active_calls import active_call_count
+
+    return ActiveCallsResponse(active_calls=active_call_count())

api/services/pipecat/active_calls.py

@@ -0,0 +1,35 @@
+"""In-process registry of active pipeline runs (live voice calls).
+
+Each uvicorn worker tracks the calls it is currently running so a deploy
+orchestrator can *drain* the worker before stopping it: poll the count, wait for
+zero, then send SIGTERM. Sending SIGTERM while calls are live makes uvicorn
+force-close their WebSockets (close code 1012), which cuts the calls instead of
+letting them finish — so the wait has to happen first.
+
+The registry is deliberately per-process. That is exactly the unit that gets
+drained: one uvicorn process per VM port (see ``scripts/rolling_update.sh``) or
+one uvicorn process per Kubernetes pod (drained via a ``preStop`` hook). The
+count is exposed read-only at ``GET /api/v1/health/active-calls`` and is also a
+natural autoscaling signal (concurrent calls per worker).
+
+Access is single-threaded (asyncio event loop), so no lock is needed. A set of
+run ids — rather than a bare counter — keeps register/unregister idempotent and
+makes the in-flight runs inspectable for debugging.
+"""
+
+_active_run_ids: set[int] = set()
+
+
+def register_active_call(workflow_run_id: int) -> None:
+    """Mark a pipeline run as active in this worker."""
+    _active_run_ids.add(workflow_run_id)
+
+
+def unregister_active_call(workflow_run_id: int) -> None:
+    """Mark a pipeline run as finished in this worker."""
+    _active_run_ids.discard(workflow_run_id)
+
+
+def active_call_count() -> int:
+    """Number of pipeline runs currently active in this worker."""
+    return len(_active_run_ids)

api/services/pipecat/run_pipeline.py

@@ -11,6 +11,10 @@ from api.services.integrations import (
     IntegrationRuntimeContext,
     create_runtime_sessions,
 )
+from api.services.pipecat.active_calls import (
+    register_active_call,
+    unregister_active_call,
+)
 from api.services.pipecat.audio_config import AudioConfig, create_audio_config
 from api.services.pipecat.event_handlers import (
     register_audio_data_handler,
@@ -870,6 +874,12 @@ async def _run_pipeline(
 
     register_audio_data_handler(audio_buffer, workflow_run_id, in_memory_audio_buffer)
 
+    # Mark this run active so a deploy can drain it before stopping this worker:
+    # the orchestrator polls GET /api/v1/health/active-calls and waits for zero
+    # before sending SIGTERM (uvicorn force-closes live call WebSockets on
+    # SIGTERM, so the wait has to come first). See api/services/pipecat/
+    # active_calls.py and scripts/rolling_update.sh.
+    register_active_call(workflow_run_id)
     try:
         # Run the pipeline
         await run_pipeline_worker(task)
@@ -877,6 +887,7 @@ async def _run_pipeline(
     except asyncio.CancelledError:
         logger.warning("Received CancelledError in _run_pipeline")
     finally:
+        unregister_active_call(workflow_run_id)
         # Close MCP sessions here, not in engine.cleanup(). The anyio cancel
         # scopes opened by MCPClient.start() in engine.initialize() are
         # task-affine; this finally runs in the same task as initialize(),

api/tests/test_active_calls.py

@@ -0,0 +1,53 @@
+"""Unit tests for the per-worker active-call registry (deploy draining).
+
+The registry backs GET /api/v1/health/active-calls, which scripts/rolling_update.sh
+(and a k8s preStop hook) polls to wait for live calls to finish before stopping a
+worker. The guarantees that matter for draining: register/unregister are
+idempotent, and the count only reaches zero when every registered run is gone.
+"""
+
+from api.services.pipecat import active_calls
+
+
+def setup_function():
+    # Module-level state — start each test from an empty registry.
+    active_calls._active_run_ids.clear()
+
+
+def test_starts_empty():
+    assert active_calls.active_call_count() == 0
+
+
+def test_register_counts_distinct_runs():
+    active_calls.register_active_call(1)
+    active_calls.register_active_call(2)
+    assert active_calls.active_call_count() == 2
+
+
+def test_register_is_idempotent():
+    # Registering the same run twice must not double-count, or the count could
+    # never drain to zero.
+    active_calls.register_active_call(1)
+    active_calls.register_active_call(1)
+    assert active_calls.active_call_count() == 1
+
+
+def test_unregister_removes_run():
+    active_calls.register_active_call(1)
+    active_calls.register_active_call(2)
+    active_calls.unregister_active_call(1)
+    assert active_calls.active_call_count() == 1
+
+
+def test_unregister_unknown_run_is_a_noop():
+    # discard() semantics: unregistering a run that was never registered (or was
+    # already removed) is safe and cannot push the count negative.
+    active_calls.unregister_active_call(999)
+    assert active_calls.active_call_count() == 0
+
+
+def test_full_lifecycle_drains_to_zero():
+    active_calls.register_active_call(42)
+    assert active_calls.active_call_count() == 1
+    active_calls.unregister_active_call(42)
+    assert active_calls.active_call_count() == 0