fix: await message confirmation interruption

backend/services/brains/base.py

@@ -95,6 +95,7 @@ class BrainRuntime:
     set_vision_scope: Callable[[dict[str, Any]], None] | None = None
     vision_function: Any = None
     set_input_enabled: Callable[[bool], None] | None = None
+    interrupt_output: Callable[[], Awaitable[None]] | None = None
     apply_turn_config: (
         Callable[[bool, dict[str, Any]], Awaitable[None]] | None
     ) = None

backend/services/brains/workflow_brain.py

@@ -19,7 +19,6 @@ from pipecat.flows import (
     NodeConfig,
 )
 from pipecat.frames.frames import (
-    InterruptionFrame,
     LLMRunFrame,
     LLMUpdateSettingsFrame,
     OutputTransportMessageUrgentFrame,
@@ -1098,10 +1097,12 @@ class WorkflowBrain(BaseBrain):
                 completion_policy == MESSAGE_CONFIRMATION
                 and result.action == "confirmed"
             ):
-                # The client-tool result closes the confirmation gate, while
+                if runtime.interrupt_output is None:
-                # InterruptionFrame closes any speech still owned by this
+                    raise RuntimeError("当前管线不支持中断 Message 确认语音")
-                # Message before the next node can enqueue new output.
+                # Keep the client-tool result as the confirmation gate, then
-                await runtime.queue_frame(InterruptionFrame())
+                # wait for the pipeline-owned interruption boundary before the
+                # next node is allowed to enqueue new output.
+                await runtime.interrupt_output()
             await self._emit_trace(
                 "message_completed",
                 nodeId=node_id,

backend/services/pipecat/pipeline.py

@@ -104,6 +104,17 @@ ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = (
     "先调用 search_knowledge_base 检索；回答资料事实时只根据检索内容，"
     "资料不足要明确说明。"
 )
+
+
+async def _interrupt_pipeline_output(
+    source: FrameProcessor,
+    worker: PipelineWorker,
+) -> None:
+    """Broadcast an interruption and wait until it crosses the media pipeline."""
+    await source.broadcast_interruption()
+    await worker.flush_pipeline(timeout=1.0)
+
+
 def _compact_knowledge_metadata(value: str, max_length: int) -> str:
     """Keep tool metadata useful without letting it dominate the model context."""
     compact = " ".join(value.split())
@@ -733,6 +744,10 @@ async def run_pipeline(
         current_enable_interrupt = enable_interrupt
         current_turn_config = normalized
 
+    async def interrupt_output() -> None:
+        """Stop active output through the same boundary as text/voice input."""
+        await _interrupt_pipeline_output(user_input, worker)
+
 
     def set_system_prompt(text: str) -> None:
         """替换上下文里的系统提示(节点切换时整体替换,而非追加)。"""
@@ -766,6 +781,7 @@ async def run_pipeline(
             set_vision_scope=lambda scope: workflow_vision_scope.update(scope),
             vision_function=workflow_vision_function,
             set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled),
+            interrupt_output=interrupt_output,
             apply_turn_config=apply_workflow_turn_config,
             flow_global_functions=flow_global_functions,
         ),

backend/tests/test_brains.py

@@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, patch
 from models import AssistantConfig, RuntimeTool
 from pipecat.flows import FlowManager
 from pipecat.frames.frames import (
-    InterruptionFrame,
     LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -1536,14 +1535,15 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
         async def queue_frame(frame):
             if isinstance(frame, TTSSpeakFrame):
                 events.append(("speech", frame.text))
-            elif isinstance(frame, InterruptionFrame):
-                events.append("interrupted")
             elif (
                 isinstance(frame, OutputTransportMessageUrgentFrame)
                 and frame.message.get("event") == "message_completed"
             ):
                 events.append("message_completed")
 
+        async def interrupt_output():
+            events.append("interrupted")
+
         class OrderedCallEnd(FakeCallEnd):
             def __init__(self):
                 super().__init__()
@@ -1580,6 +1580,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
             call_end=call_end,
             client_tools=client_tools,
             set_input_enabled=input_states.append,
+            interrupt_output=interrupt_output,
         )
         brain._message_stages.set_client_tools(client_tools)
 

backend/tests/test_pipeline_events.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import unittest
 from types import SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
@@ -10,6 +10,7 @@ from pipecat.frames.frames import (
     OutputTransportMessageUrgentFrame,
 )
 from services.pipecat.pipeline_events import bind_cascade_pipeline_events
+from services.pipecat.pipeline import _interrupt_pipeline_output
 
 
 class _EventSource:
@@ -80,6 +81,25 @@ class _Brain:
 
 
 class PipelineEventTest(unittest.IsolatedAsyncioTestCase):
+    async def test_output_interruption_broadcasts_before_flush_barrier(self):
+        events = []
+        source = SimpleNamespace(
+            broadcast_interruption=AsyncMock(
+                side_effect=lambda: events.append("broadcast")
+            )
+        )
+        worker = SimpleNamespace(
+            flush_pipeline=AsyncMock(
+                side_effect=lambda **_kwargs: events.append("flush")
+            )
+        )
+
+        await _interrupt_pipeline_output(source, worker)
+
+        self.assertEqual(events, ["broadcast", "flush"])
+        source.broadcast_interruption.assert_awaited_once_with()
+        worker.flush_pipeline.assert_awaited_once_with(timeout=1.0)
+
     async def test_greeting_keeps_playback_timestamp_until_client_ready(self):
         transport = _EventSource()
         text_input = _EventSource()