fix: harden realtime pipeline lifecycle

backend/requirements.txt

@@ -18,6 +18,8 @@ python-dotenv
 pydantic
 loguru
 websockets>=13
+# websockets uses the system proxy; SOCKS connections need this optional extra.
+python-socks[asyncio]>=2.4,<3
 
 # 存储:Postgres(SQLAlchemy 2.0 异步 + asyncpg 驱动)
 sqlalchemy[asyncio]>=2.0

backend/routes/voice_webrtc.py

@@ -145,7 +145,10 @@ async def http_offer(request: Request, payload: dict = Body(...)):
         # Client capabilities have one canonical transport: the request header.
         "client_tool_names": client_tool_names,
     }
-    answer = await _handle_offer_payload(offer_payload, _http_peers)
+    try:
+        answer = await _handle_offer_payload(offer_payload, _http_peers)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
     return answer
 
 

backend/services/config_resolver.py

@@ -34,7 +34,7 @@ async def _resource_for(
     binding = await session.get(AssistantModelBinding, (assistant_id, capability))
     resource_id = binding.model_resource_id if binding else None
     resource = await session.get(ModelResource, resource_id) if resource_id else None
-    if resource and resource.capability != capability:
+    if resource and (not resource.enabled or resource.capability != capability):
         resource = None
     if resource is None:
         resource = (
@@ -60,6 +60,7 @@ async def _agent_resource_for(
     resource = await session.get(ModelResource, resource_id) if resource_id else None
     if (
         resource
+        and resource.enabled
         and resource.capability == "Agent"
         and resource.interface_type == interface_type
     ):

backend/services/conversation_history.py

@@ -16,6 +16,8 @@ from services.object_storage import delete_object, put_object, storage_uri
 
 
 MAX_WORKFLOW_TRACE_EVENTS = 1000
+MAX_PENDING_HISTORY_MESSAGES = 256
+HISTORY_FLUSH_TIMEOUT_SECONDS = 3.0
 
 
 def _parse_timestamp(value: object) -> datetime:
@@ -39,6 +41,63 @@ class ConversationRecorder:
         self._lock = asyncio.Lock()
         self._seen_events: set[str] = set()
         self._pending_artifacts: set[asyncio.Task[None]] = set()
+        self._pending_messages: asyncio.Queue[dict] = asyncio.Queue(
+            maxsize=MAX_PENDING_HISTORY_MESSAGES,
+        )
+        self._message_writer: asyncio.Task[None] | None = None
+        self._accepting_messages = True
+
+    def record_transport_message_later(self, message: object) -> None:
+        """Keep database latency off the media path, with bounded buffering."""
+        if (
+            not self._accepting_messages
+            or not isinstance(message, dict)
+            or message.get("type") not in {
+                "transcript", "assistant-text-end", "workflow-event",
+            }
+        ):
+            return
+        try:
+            self._pending_messages.put_nowait(deepcopy(message))
+        except asyncio.QueueFull:
+            logger.warning(f"对话历史写入积压，跳过本条记录: session={self.session_id}")
+            return
+        if self._message_writer is None:
+            self._message_writer = asyncio.create_task(
+                self._write_pending_messages(),
+                name=f"conversation-history-writer:{self.session_id}",
+            )
+
+    async def _write_pending_messages(self) -> None:
+        while True:
+            message = await self._pending_messages.get()
+            try:
+                await self.record_transport_message(message)
+            except Exception as exc:
+                logger.warning(f"对话历史写入失败: {type(exc).__name__}")
+            finally:
+                self._pending_messages.task_done()
+
+    async def flush_pending_messages(self) -> None:
+        """Drain final transcripts before finishing the conversation row."""
+        self._accepting_messages = False
+        writer = self._message_writer
+        if writer is None:
+            return
+        try:
+            await asyncio.wait_for(
+                self._pending_messages.join(),
+                timeout=HISTORY_FLUSH_TIMEOUT_SECONDS,
+            )
+        except TimeoutError:
+            logger.warning(f"对话历史收尾超时，丢弃剩余缓冲: session={self.session_id}")
+        finally:
+            writer.cancel()
+            await asyncio.gather(writer, return_exceptions=True)
+            self._message_writer = None
+            while not self._pending_messages.empty():
+                self._pending_messages.get_nowait()
+                self._pending_messages.task_done()
 
     @classmethod
     async def start(
@@ -288,6 +347,7 @@ class ConversationRecorder:
             raise
 
     async def _finish(self, *, status: str) -> None:
+        await self.flush_pending_messages()
         pending_artifacts = list(self._pending_artifacts)
         if pending_artifacts:
             await asyncio.gather(*pending_artifacts, return_exceptions=True)

backend/services/pipecat/pipeline.py

@@ -103,7 +103,7 @@ from services.realtime.protocol import (
     PipelineProtocolRuntime,
     RealtimeProviderControlFrame,
 )
-from pipecat.workers.runner import WorkerRunner
+from services.pipecat.pipeline_runner import run_media_worker
 
 
 KNOWLEDGE_TOOL_NAME = "search_knowledge_base"
@@ -260,6 +260,7 @@ async def run_pipeline(
             assistant_id=assistant_id,
             channel=channel,
             protocol_adapter=protocol_adapter,
+            client_tool_names=client_tool_names,
         )
         return
 
@@ -987,25 +988,7 @@ async def run_pipeline(
         submit_user_input=submit_user_input,
         client_tools=client_tools,
     )
-    runner = WorkerRunner(handle_sigint=False)
-    run_status = "completed"
-    can_finalize_recorder = True
-    try:
-        await runner.add_workers(worker)
-        await runner.run()
-    except GeneratorExit:
-        # A coroutine being synchronously closed cannot perform async cleanup.
-        # The WebRTC route now prevents this by retaining every pipeline task;
-        # keep this guard so a future unmanaged caller cannot corrupt asyncpg.
-        run_status = "failed"
-        can_finalize_recorder = False
-        raise
-    except Exception:
-        run_status = "failed"
-        raise
-    finally:
-        if recorder and can_finalize_recorder:
-            await recorder.finish(status=run_status)
+    await run_media_worker(worker, recorder)
     logger.info("管线已结束")
 
 
@@ -1018,6 +1001,7 @@ async def run_realtime_pipeline(
     assistant_id: str | None = None,
     channel: str = "webrtc",
     protocol_adapter: PipelineProtocolAdapter | None = None,
+    client_tool_names: set[str] | None = None,
 ) -> None:
     """Run a speech-to-speech model that owns ASR, reasoning, and synthesis."""
     realtime = create_realtime_service(
@@ -1239,20 +1223,5 @@ async def run_realtime_pipeline(
         vision_enabled=vision_enabled,
         vision_state=vision_state,
     )
-    runner = WorkerRunner(handle_sigint=False)
-    run_status = "completed"
-    can_finalize_recorder = True
-    try:
-        await runner.add_workers(worker)
-        await runner.run()
-    except GeneratorExit:
-        run_status = "failed"
-        can_finalize_recorder = False
-        raise
-    except Exception:
-        run_status = "failed"
-        raise
-    finally:
-        if recorder and can_finalize_recorder:
-            await recorder.finish(status=run_status)
+    await run_media_worker(worker, recorder)
     logger.info("Realtime 管线已结束")

backend/services/pipecat/pipeline_runner.py

@@ -0,0 +1,56 @@
+"""Run a media worker and persist its actual terminal outcome."""
+
+import asyncio
+
+from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame
+from pipecat.pipeline.worker import PipelineWorker
+from pipecat.workers.runner import WorkerRunner
+
+from services.conversation_history import ConversationRecorder
+
+
+async def run_media_worker(
+    worker: PipelineWorker,
+    recorder: ConversationRecorder | None,
+) -> None:
+    # A runner can return normally after a fatal ErrorFrame or a failed worker.
+    # Only an observed, expected terminal frame counts as normal completion.
+    status = "failed"
+    fatal_error = False
+
+    @worker.event_handler("on_pipeline_error")
+    async def on_pipeline_error(_worker, frame: ErrorFrame):
+        nonlocal status, fatal_error
+        if frame.fatal:
+            fatal_error = True
+            status = "failed"
+
+    @worker.event_handler("on_pipeline_finished")
+    async def on_pipeline_finished(_worker, frame):
+        nonlocal status
+        normal_end = isinstance(frame, EndFrame) or (
+            isinstance(frame, CancelFrame)
+            and frame.reason in {"client_disconnected", "idle timeout"}
+        )
+        if normal_end and not fatal_error:
+            status = "completed"
+
+    runner = WorkerRunner(handle_sigint=False)
+    can_finalize = True
+    try:
+        await runner.add_workers(worker)
+        await runner.run()
+        # Pipecat's runner consumes cancellation while draining its workers.
+        # Preserve the owning server task's cancellation after that cleanup.
+        owner = asyncio.current_task()
+        if owner is not None and owner.cancelling():
+            raise asyncio.CancelledError
+    except GeneratorExit:
+        can_finalize = False  # Synchronously closed coroutines cannot await.
+        raise
+    except BaseException:
+        status = "failed"
+        raise
+    finally:
+        if recorder and can_finalize:
+            await recorder.finish(status=status)

backend/services/pipecat/processors.py

@@ -657,7 +657,7 @@ class ConversationHistoryProcessor(FrameProcessor):
         await super().process_frame(frame, direction)
         await self.push_frame(frame, direction)
         if self._recorder and isinstance(frame, OutputTransportMessageUrgentFrame):
-            await self._recorder.record_transport_message(frame.message)
+            self._recorder.record_transport_message_later(frame.message)
 
 
 class KnowledgeRetrievalProcessor(FrameProcessor):

backend/services/realtime/lifecycle.py

@@ -12,13 +12,33 @@ from loguru import logger
 active_connections: set[object] = set()
 pipeline_tasks: set[asyncio.Task[None]] = set()
 connection_tasks: dict[object, asyncio.Task[None]] = {}
+connection_cleanup_tasks: set[asyncio.Task[None]] = set()
 DEFAULT_CLOSE_GRACE_SECONDS = 10.0
 
 
+async def _disconnect_connection(connection: object) -> None:
+    """Release media even when construction failed before transport startup."""
+    try:
+        async with asyncio.timeout(DEFAULT_CLOSE_GRACE_SECONDS):
+            await connection.disconnect()
+    except Exception as exc:
+        logger.warning(f"Realtime 连接清理失败: {type(exc).__name__}")
+
+
 def _consume_pipeline_result(task: asyncio.Task[None], connection: object) -> None:
     pipeline_tasks.discard(task)
     if connection_tasks.get(connection) is task:
         connection_tasks.pop(connection, None)
+    if connection in active_connections:
+        active_connections.discard(connection)
+        # Run after the pipeline task is done: closed handlers can safely wait
+        # for that task without making pipeline shutdown wait on itself.
+        cleanup = asyncio.create_task(
+            _disconnect_connection(connection),
+            name=f"realtime-disconnect:{getattr(connection, 'pc_id', id(connection))}",
+        )
+        connection_cleanup_tasks.add(cleanup)
+        cleanup.add_done_callback(connection_cleanup_tasks.discard)
     try:
         error = task.exception()
     except asyncio.CancelledError:
@@ -86,11 +106,15 @@ async def shutdown_active_sessions(
 ) -> None:
     connections = list(active_connections)
     if connections:
+        active_connections.difference_update(connections)
         await asyncio.gather(
-            *(connection.disconnect() for connection in connections),
+            *(_disconnect_connection(connection) for connection in connections),
             return_exceptions=True,
         )
-        active_connections.difference_update(connections)
+
+    cleanup_tasks = list(connection_cleanup_tasks)
+    if cleanup_tasks:
+        await asyncio.gather(*cleanup_tasks, return_exceptions=True)
 
     tasks = list(pipeline_tasks)
     if not tasks:
@@ -106,4 +130,3 @@ async def shutdown_active_sessions(
     logger.info(
         f"Realtime 会话清理完成: normal={len(done)} cancelled={len(cancelled)}"
     )
-

backend/tests/test_config_resolver.py

@@ -0,0 +1,38 @@
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+from services.config_resolver import _agent_resource_for, _resource_for
+
+
+class DisabledResourceTest(unittest.IsolatedAsyncioTestCase):
+    async def test_disabled_binding_uses_enabled_fallback_for_models_and_agents(self):
+        for capability, interface_type in (("LLM", "openai-llm"), ("Agent", "dify")):
+            with self.subTest(capability=capability):
+                disabled = SimpleNamespace(
+                    id="disabled", capability=capability,
+                    interface_type=interface_type, enabled=False,
+                )
+                fallback = SimpleNamespace(id="enabled", enabled=True)
+                result = Mock()
+                result.scalar_one_or_none.return_value = fallback
+                session = SimpleNamespace(
+                    get=AsyncMock(side_effect=[SimpleNamespace(model_resource_id="disabled"), disabled]),
+                    execute=AsyncMock(return_value=result),
+                )
+                if capability == "Agent":
+                    selected = await _agent_resource_for(session, "assistant", interface_type)
+                else:
+                    selected = await _resource_for(session, "assistant", capability)
+                self.assertIs(selected, fallback)
+                statement = str(session.execute.await_args.args[0])
+                self.assertIn("model_resources.enabled IS true", statement)
+
+    async def test_enabled_binding_is_preserved(self):
+        resource = SimpleNamespace(id="enabled", capability="LLM", enabled=True)
+        session = SimpleNamespace(
+            get=AsyncMock(side_effect=[SimpleNamespace(model_resource_id=resource.id), resource]),
+            execute=AsyncMock(),
+        )
+        self.assertIs(await _resource_for(session, "assistant", "LLM"), resource)
+        session.execute.assert_not_awaited()

backend/tests/test_conversation_history.py

@@ -9,6 +9,91 @@ from services.conversation_history import ConversationRecorder
 
 
 class ConversationRecorderTest(unittest.IsolatedAsyncioTestCase):
+    async def test_background_messages_keep_order_and_ignore_streaming_deltas(self):
+        recorder = ConversationRecorder("conv_test")
+        recorder.record_transport_message = AsyncMock()
+        events = [
+            {"type": "transcript", "content": "user"},
+            {"type": "assistant-text-end", "content": "assistant"},
+        ]
+        for event in events:
+            recorder.record_transport_message_later(event)
+        recorder.record_transport_message_later({"type": "assistant-text-delta"})
+        await recorder.flush_pending_messages()
+        self.assertEqual(
+            [call.args[0] for call in recorder.record_transport_message.await_args_list],
+            events,
+        )
+        self.assertIsNone(recorder._message_writer)
+
+    async def test_slow_database_does_not_block_audio(self):
+        from pipecat.frames.frames import EndFrame, OutputAudioRawFrame, OutputTransportMessageUrgentFrame
+        from pipecat.pipeline.pipeline import Pipeline
+        from pipecat.pipeline.worker import PipelineWorker
+        from pipecat.processors.frame_processor import FrameProcessor
+        from pipecat.workers.runner import WorkerRunner
+        from services.pipecat.processors import ConversationHistoryProcessor
+
+        entered, release, audio_received = asyncio.Event(), asyncio.Event(), asyncio.Event()
+        recorder = ConversationRecorder("conv_test")
+
+        async def slow_write(message):
+            entered.set()
+            await release.wait()
+
+        recorder.record_transport_message = slow_write
+
+        class AudioSink(FrameProcessor):
+            async def process_frame(self, frame, direction):
+                await super().process_frame(frame, direction)
+                if isinstance(frame, OutputAudioRawFrame):
+                    audio_received.set()
+                await self.push_frame(frame, direction)
+
+        worker = PipelineWorker(
+            Pipeline([ConversationHistoryProcessor(recorder), AudioSink()]), enable_rtvi=False,
+        )
+        runner = WorkerRunner(handle_sigint=False)
+        await runner.add_workers(worker)
+        task = asyncio.create_task(runner.run())
+        try:
+            await worker.queue_frame(OutputTransportMessageUrgentFrame(message={"type": "transcript"}))
+            await asyncio.wait_for(entered.wait(), 3)
+            await worker.queue_frame(OutputAudioRawFrame(audio=b"\0" * 320, sample_rate=16000, num_channels=1))
+            await asyncio.wait_for(audio_received.wait(), 1)
+            self.assertFalse(release.is_set())
+        finally:
+            release.set()
+            await recorder.flush_pending_messages()
+            await worker.queue_frame(EndFrame())
+            await asyncio.wait_for(task, 3)
+
+    async def test_history_buffer_is_bounded_and_slow_writer_is_cancelled(self):
+        started = asyncio.Event()
+        cancelled = asyncio.Event()
+        with patch("services.conversation_history.MAX_PENDING_HISTORY_MESSAGES", 2):
+            recorder = ConversationRecorder("conv_test")
+
+        async def blocked_write(message):
+            started.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                cancelled.set()
+
+        recorder.record_transport_message = blocked_write
+        recorder.record_transport_message_later({"type": "transcript"})
+        await started.wait()
+        for _ in range(4):
+            recorder.record_transport_message_later({"type": "transcript"})
+        self.assertEqual(recorder._pending_messages.qsize(), 2)
+        with patch("services.conversation_history.HISTORY_FLUSH_TIMEOUT_SECONDS", 0.01):
+            await recorder.flush_pending_messages()
+        self.assertTrue(cancelled.is_set())
+        self.assertTrue(recorder._pending_messages.empty())
+        recorder.record_transport_message_later({"type": "transcript"})
+        self.assertIsNone(recorder._message_writer)
+
     async def test_completed_conversation_queues_enabled_analysis(self):
         conversation = SimpleNamespace(
             status="active",

backend/tests/test_pipeline_runtime.py

@@ -0,0 +1,101 @@
+"""Regression checks for whole-pipeline assembly and terminal outcomes."""
+
+import asyncio
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from models import AssistantConfig
+from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, StartFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.worker import PipelineWorker
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from services.pipecat import pipeline
+from services.pipecat.pipeline_runner import run_media_worker
+
+
+class PipelineAssemblyTest(unittest.IsolatedAsyncioTestCase):
+    async def test_realtime_startup_preserves_client_capabilities(self):
+        brain = SimpleNamespace(
+            spec=SimpleNamespace(supported_runtime_modes={"realtime"}),
+            system_prompt=lambda cfg: "test",
+            greeting=AsyncMock(return_value=""),
+            setup_realtime=AsyncMock(),
+            on_session_update=AsyncMock(),
+        )
+        transport = SimpleNamespace(
+            input=lambda: FrameProcessor(),
+            output=lambda: FrameProcessor(),
+            event_handler=lambda name: lambda handler: handler,
+        )
+        with (
+            patch.object(pipeline, "build_brain", return_value=brain),
+            patch.object(pipeline, "create_realtime_service", return_value=FrameProcessor()),
+            patch.object(pipeline.ConversationRecorder, "start", new=AsyncMock(return_value=None)),
+            patch.object(pipeline, "run_media_worker", new=AsyncMock()) as run,
+        ):
+            await pipeline.run_pipeline(
+                transport, AssistantConfig(runtimeMode="realtime"),
+                client_tool_names={"show_message"},
+            )
+        run.assert_awaited_once()
+        runtime = brain.setup_realtime.await_args.args[1]
+        self.assertTrue(runtime.client_tools.supports("show_message"))
+        self.assertFalse(runtime.client_tools.supports("unregistered_tool"))
+
+
+class TerminalProcessor(FrameProcessor):
+    def __init__(self, terminal):
+        super().__init__()
+        self.terminal = terminal
+
+    async def process_frame(self, frame, direction):
+        await super().process_frame(frame, direction)
+        await self.push_frame(frame, direction)
+        if isinstance(frame, StartFrame):
+            if isinstance(self.terminal, ErrorFrame):
+                await self.push_frame(self.terminal, FrameDirection.UPSTREAM)
+            else:
+                await self.push_frame(self.terminal)
+
+
+class PipelineOutcomeTest(unittest.IsolatedAsyncioTestCase):
+    async def test_fatal_error_is_failed_even_when_runner_returns_normally(self):
+        worker = PipelineWorker(
+            Pipeline([TerminalProcessor(ErrorFrame(error="provider failed", fatal=True))]),
+            enable_rtvi=False,
+        )
+        recorder = SimpleNamespace(finish=AsyncMock())
+        await asyncio.wait_for(run_media_worker(worker, recorder), 3)
+        recorder.finish.assert_awaited_once_with(status="failed")
+
+    async def test_normal_end_and_client_disconnect_are_completed(self):
+        for terminal in (EndFrame(), CancelFrame(reason="client_disconnected")):
+            with self.subTest(terminal=type(terminal).__name__):
+                worker = PipelineWorker(Pipeline([Passthrough()]), enable_rtvi=False)
+                recorder = SimpleNamespace(finish=AsyncMock())
+                await worker.queue_frame(terminal)
+                await asyncio.wait_for(run_media_worker(worker, recorder), 3)
+                recorder.finish.assert_awaited_once_with(status="completed")
+
+    async def test_server_task_cancellation_remains_failed_and_propagates(self):
+        worker = PipelineWorker(Pipeline([Passthrough()]), enable_rtvi=False)
+        started = asyncio.Event()
+
+        @worker.event_handler("on_pipeline_started")
+        async def on_started(_worker, _frame):
+            started.set()
+
+        recorder = SimpleNamespace(finish=AsyncMock())
+        task = asyncio.create_task(run_media_worker(worker, recorder))
+        await asyncio.wait_for(started.wait(), 3)
+        task.cancel()
+        with self.assertRaises(asyncio.CancelledError):
+            await asyncio.wait_for(task, 3)
+        recorder.finish.assert_awaited_once_with(status="failed")
+
+
+class Passthrough(FrameProcessor):
+    async def process_frame(self, frame, direction):
+        await super().process_frame(frame, direction)
+        await self.push_frame(frame, direction)

backend/tests/test_voice_webrtc_lifecycle.py

@@ -8,12 +8,16 @@ from unittest.mock import AsyncMock, patch
 
 from fastapi import HTTPException
 from routes import voice_webrtc
+from services.realtime import lifecycle
 from starlette.requests import Request
 
 
 class FakeConnection:
     pc_id = "pc_test"
 
+    def __init__(self):
+        self.disconnect = AsyncMock()
+
 
 def encoded_header_json(value: object) -> str:
     payload = json.dumps(value).encode("utf-8")
@@ -81,6 +85,16 @@ class WebRTCOfferMetadataTest(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual(raised.exception.status_code, 400)
 
+    async def test_invalid_assistant_returns_actionable_bad_request(self):
+        with patch.object(
+            voice_webrtc, "_handle_offer_payload",
+            new=AsyncMock(side_effect=ValueError("助手不存在: missing")),
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await voice_webrtc.http_offer(offer_request(), {"type": "offer"})
+        self.assertEqual(raised.exception.status_code, 400)
+        self.assertIn("助手不存在", raised.exception.detail)
+
 
 class WebRTCPipelineLifecycleTest(unittest.IsolatedAsyncioTestCase):
     async def asyncTearDown(self):
@@ -89,6 +103,9 @@ class WebRTCPipelineLifecycleTest(unittest.IsolatedAsyncioTestCase):
             task.cancel()
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
+        await asyncio.sleep(0)
+        if lifecycle.connection_cleanup_tasks:
+            await asyncio.gather(*list(lifecycle.connection_cleanup_tasks))
         voice_webrtc._pipeline_tasks.clear()
         voice_webrtc._connection_tasks.clear()
         voice_webrtc._active_connections.clear()
@@ -111,6 +128,54 @@ class WebRTCPipelineLifecycleTest(unittest.IsolatedAsyncioTestCase):
         self.assertNotIn(task, voice_webrtc._pipeline_tasks)
         self.assertNotIn(connection, voice_webrtc._connection_tasks)
 
+    async def test_failed_startup_disconnects_and_releases_connection(self):
+        connection = FakeConnection()
+
+        async def pipeline():
+            raise ValueError("provider construction failed")
+
+        task = voice_webrtc._start_pipeline_task(connection, pipeline())
+        with self.assertRaises(ValueError):
+            await task
+        await asyncio.sleep(0)
+        if lifecycle.connection_cleanup_tasks:
+            await asyncio.gather(*list(lifecycle.connection_cleanup_tasks))
+        connection.disconnect.assert_awaited_once()
+        self.assertNotIn(connection, voice_webrtc._active_connections)
+        self.assertNotIn(connection, voice_webrtc._connection_tasks)
+
+    async def test_cleanup_closed_handler_can_wait_for_completed_pipeline(self):
+        connection = FakeConnection()
+
+        async def pipeline():
+            return None
+
+        task = voice_webrtc._start_pipeline_task(connection, pipeline())
+
+        async def disconnect():
+            await lifecycle.wait_for_pipeline_close(
+                task, connection_id=connection.pc_id, timeout=0.01,
+            )
+
+        connection.disconnect.side_effect = disconnect
+        await task
+        await asyncio.sleep(0)
+        if lifecycle.connection_cleanup_tasks:
+            await asyncio.wait_for(
+                asyncio.gather(*list(lifecycle.connection_cleanup_tasks)), 0.5,
+            )
+        connection.disconnect.assert_awaited_once()
+
+    async def test_already_closed_connection_is_not_disconnected_twice(self):
+        connection = FakeConnection()
+        release = asyncio.Event()
+        task = voice_webrtc._start_pipeline_task(connection, release.wait())
+        voice_webrtc._active_connections.discard(connection)
+        release.set()
+        await task
+        await asyncio.sleep(0)
+        connection.disconnect.assert_not_awaited()
+
     async def test_stuck_pipeline_is_cancelled_after_close_grace_period(self):
         connection = FakeConnection()
         cancelled = asyncio.Event()