fix: relay confirmation interrupts past muted input

backend/services/brains/base.py

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

backend/services/brains/workflow_brain.py

@@ -1093,16 +1093,6 @@ class WorkflowBrain(BaseBrain):
             ),
         )
         if result.succeeded:
-            if (
-                completion_policy == MESSAGE_CONFIRMATION
-                and result.action == "confirmed"
-            ):
-                if runtime.wait_for_output_stopped is None:
-                    raise RuntimeError("当前管线不支持等待 Message 确认语音停止")
-                # ClientToolBroker has already broadcast the interruption
-                # before resolving the confirmation. Keep the next node behind
-                # the pipeline and transport playback boundaries.
-                await runtime.wait_for_output_stopped()
             await self._emit_trace(
                 "message_completed",
                 nodeId=node_id,

backend/services/client_tools.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+from collections import deque
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from typing import Any, Literal, Protocol
 
@@ -30,6 +32,14 @@ class _PendingClientToolCall:
     interrupt_on_result: bool = False
 
 
+@dataclass(frozen=True)
+class _DeferredClientToolResult:
+    """A successful result waiting for the media interruption barrier."""
+
+    future: asyncio.Future[dict[str, Any]]
+    response: dict[str, Any]
+
+
 class ClientToolPort(Protocol):
     async def call(
         self,
@@ -49,8 +59,17 @@ class ClientToolBroker(FrameProcessor):
     def __init__(self) -> None:
         super().__init__()
         self._pending: dict[str, _PendingClientToolCall] = {}
+        self._deferred_results: deque[_DeferredClientToolResult] = deque()
+        self._interrupt_handler: Callable[[], Awaitable[None]] | None = None
         self._closed_message: str | None = None
 
+    def set_interrupt_handler(
+        self,
+        handler: Callable[[], Awaitable[None]],
+    ) -> None:
+        """Use a processor downstream of muted user input as interrupt source."""
+        self._interrupt_handler = handler
+
     async def call(
         self,
         function_name: str,
@@ -142,21 +161,42 @@ class ClientToolBroker(FrameProcessor):
             logger.debug(f"忽略未知或过期的客户端工具结果: {tool_call_id}")
             return
 
-        status = str(message.get("status") or "error")
-        # Keep interruption in the input-frame path, matching text input.
-        # Resolve the tool future only after the interruption has been
-        # broadcast, so the workflow cannot enqueue its next node first.
-        if status == "ok" and pending.interrupt_on_result:
-            await self.broadcast_interruption()
-
         future = pending.future
+        status = str(message.get("status") or "error")
         if status == "ok":
-            future.set_result(
+            response = {
-                {
                 "status": "ok",
                 "data": message.get("data"),
             }
+            if pending.interrupt_on_result:
+                # Match text input exactly: broadcasting only starts the
+                # interruption. The assistant aggregator calls
+                # on_interruption_processed() after the old response state has
+                # actually been cleared; only then may the workflow continue.
+                deferred = _DeferredClientToolResult(
+                    future=future,
+                    response=response,
                 )
+                self._deferred_results.append(deferred)
+                try:
+                    if self._interrupt_handler is not None:
+                        await self._interrupt_handler()
+                    else:
+                        await self.broadcast_interruption()
+                except Exception as exc:
+                    try:
+                        self._deferred_results.remove(deferred)
+                    except ValueError:
+                        # The downstream acknowledgement may have won the
+                        # race before an upstream broadcast failure arrived.
+                        pass
+                    if not future.done():
+                        future.set_exception(
+                            ClientToolError("客户端工具结果中断处理失败")
+                        )
+                    logger.exception(f"客户端工具结果中断失败: {exc}")
+                return
+            future.set_result(response)
         else:
             future.set_result(
                 {
@@ -166,7 +206,18 @@ class ClientToolBroker(FrameProcessor):
                 }
             )
 
+    def on_interruption_processed(self) -> None:
+        """Release one result after the aggregator acknowledges interruption."""
+        while self._deferred_results:
+            deferred = self._deferred_results.popleft()
+            if deferred.future.done():
+                continue
+            deferred.future.set_result(deferred.response)
+            logger.debug("客户端工具结果已通过中断处理边界")
+            return
+
     def _fail_pending(self, message: str) -> None:
+        self._deferred_results.clear()
         for pending in self._pending.values():
             future = pending.future
             if not future.done():

backend/services/pipecat/pipeline.py

@@ -8,7 +8,6 @@
 
 import asyncio
 import base64
-from collections.abc import Awaitable, Callable
 from io import BytesIO
 from typing import Any
 
@@ -107,21 +106,6 @@ ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = (
 )
 
 
-async def _wait_for_output_stop(
-    wait_until_stopped: Callable[[], Awaitable[None]],
-    *,
-    timeout_seconds: float = 3.0,
-) -> None:
-    """Wait for the transport's real stop event after an interruption."""
-    try:
-        await asyncio.wait_for(
-            wait_until_stopped(),
-            timeout=timeout_seconds,
-        )
-    except TimeoutError as exc:
-        raise RuntimeError("等待客户端语音停止超时") from exc
-
-
 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())
@@ -437,6 +421,11 @@ async def run_pipeline(
         should_ignore_update=lambda: call_end.ending,
     )
     client_tools = ClientToolBroker()
+    # Confirmation nodes mute the user aggregator, which intentionally drops
+    # InterruptionFrame. Relay confirmation interruptions from the next
+    # processor instead so LLM, assistant aggregation, TTS, and output all see
+    # the frame before the workflow continues.
+    client_tools.set_interrupt_handler(user_turn_router.broadcast_interruption)
     vision_capture = VisionCaptureProcessor()
     knowledge_retrieval = KnowledgeRetrievalProcessor(
         automatic_knowledge_id,
@@ -751,11 +740,6 @@ async def run_pipeline(
         current_enable_interrupt = enable_interrupt
         current_turn_config = normalized
 
-    async def wait_for_output_stopped() -> None:
-        """Keep workflow continuation behind the interrupted output."""
-        if call_end.speaking:
-            await _wait_for_output_stop(call_end.wait_until_silent)
-
     def set_system_prompt(text: str) -> None:
         """替换上下文里的系统提示(节点切换时整体替换,而非追加)。"""
         messages = context.get_messages()
@@ -788,7 +772,6 @@ 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),
-            wait_for_output_stopped=wait_for_output_stopped,
             apply_turn_config=apply_workflow_turn_config,
             flow_global_functions=flow_global_functions,
         ),
@@ -877,6 +860,7 @@ async def run_pipeline(
         vision_enabled=vision_enabled,
         vision_state=vision_state,
         submit_user_input=submit_user_input,
+        client_tools=client_tools,
     )
     runner = WorkerRunner(handle_sigint=False)
     run_status = "completed"

backend/services/pipecat/pipeline_events.py

@@ -16,6 +16,7 @@ from pipecat.runner.utils import (
     maybe_capture_participant_camera,
 )
 from pipecat.utils.time import time_now_iso8601
+from services.client_tools import ClientToolBroker
 from services.pipecat.processors import UserInput
 
 
@@ -32,6 +33,7 @@ def bind_cascade_pipeline_events(
     vision_enabled: bool,
     vision_state: dict[str, str | None],
     submit_user_input: Callable[[UserInput], Awaitable[None]] | None = None,
+    client_tools: ClientToolBroker | None = None,
 ) -> None:
     """Connect processors to transport events without owning pipeline assembly."""
 
@@ -159,6 +161,8 @@ def bind_cascade_pipeline_events(
 
     @assistant_aggregator.event_handler("on_interruption_processed")
     async def on_interruption_processed(_aggregator):
+        if client_tools is not None:
+            client_tools.on_interruption_processed()
         if not pending_user_inputs:
             return
         await finish_user_input(pending_user_inputs.pop(0))

backend/tests/test_brains.py

@@ -1541,9 +1541,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
             ):
                 events.append("message_completed")
 
-        async def wait_for_output_stopped():
-            events.append("output_stopped")
-
         class OrderedCallEnd(FakeCallEnd):
             def __init__(self):
                 super().__init__()
@@ -1580,7 +1577,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
             call_end=call_end,
             client_tools=client_tools,
             set_input_enabled=input_states.append,
-            wait_for_output_stopped=wait_for_output_stopped,
         )
         brain._message_stages.set_client_tools(client_tools)
 
@@ -1606,7 +1602,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(result.action, "confirmed")
         self.assertFalse(call_end.playback_completion.done())
         self.assertEqual(input_states, [False, True])
-        self.assertEqual(events[-2:], ["output_stopped", "message_completed"])
+        self.assertEqual(events[-1], "message_completed")
 
     async def test_speech_only_message_waits_for_transport_playback(self):
         brain = WorkflowBrain(

backend/tests/test_client_tools.py

@@ -138,7 +138,52 @@ class ClientToolExecutorTests(unittest.IsolatedAsyncioTestCase):
 
 
 class ClientToolBrokerTests(unittest.IsolatedAsyncioTestCase):
-    async def test_interrupts_before_resolving_configured_result(self):
+    async def test_uses_configured_downstream_interrupt_handler(self):
+        broker = ClientToolBroker()
+        outbound = []
+        interruptions = []
+
+        async def push_frame(frame, direction=FrameDirection.DOWNSTREAM):
+            outbound.append((frame, direction))
+
+        async def interrupt_after_muted_input():
+            interruptions.append("downstream")
+
+        broker.push_frame = push_frame
+        broker.set_interrupt_handler(interrupt_after_muted_input)
+        call = asyncio.create_task(
+            broker.call(
+                "show_message",
+                {},
+                timeout_seconds=1,
+                response_wait_mode="session",
+                interrupt_on_result=True,
+            )
+        )
+        await asyncio.sleep(0)
+        message = outbound[0][0].message
+
+        await broker.process_frame(
+            InputTransportMessageFrame(
+                message={
+                    "type": "client-tool-result",
+                    "tool_call_id": message["tool_call_id"],
+                    "status": "ok",
+                    "data": {"action": "confirmed"},
+                }
+            ),
+            FrameDirection.DOWNSTREAM,
+        )
+
+        self.assertEqual(interruptions, ["downstream"])
+        self.assertFalse(call.done())
+        broker.on_interruption_processed()
+        self.assertEqual(
+            await call,
+            {"status": "ok", "data": {"action": "confirmed"}},
+        )
+
+    async def test_resolves_configured_result_after_interruption_processed(self):
         broker = ClientToolBroker()
         outbound = []
         observed_future_states = []
@@ -178,6 +223,10 @@ class ClientToolBrokerTests(unittest.IsolatedAsyncioTestCase):
         )
 
         self.assertEqual(observed_future_states, [[False]])
+        self.assertFalse(call.done())
+
+        broker.on_interruption_processed()
+
         self.assertEqual(
             await call,
             {"status": "ok", "data": {"action": "confirmed"}},

backend/tests/test_pipeline_events.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import unittest
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
@@ -11,7 +10,6 @@ from pipecat.frames.frames import (
     OutputTransportMessageUrgentFrame,
 )
 from services.pipecat.pipeline_events import bind_cascade_pipeline_events
-from services.pipecat.pipeline import _wait_for_output_stop
 
 
 class _EventSource:
@@ -82,26 +80,38 @@ class _Brain:
 
 
 class PipelineEventTest(unittest.IsolatedAsyncioTestCase):
-    async def test_interrupted_output_waits_for_transport_stop(self):
+    async def test_interruption_acknowledges_deferred_client_tool_result(self):
-        events = []
+        transport = _EventSource()
-
+        text_input = _EventSource()
-        async def wait_until_stopped():
+        user_aggregator = _EventSource()
-            events.append("stopped")
+        assistant_aggregator = _EventSource()
-
+        worker = _Worker()
-        await _wait_for_output_stop(wait_until_stopped)
+        brain = _Brain(worker)
-
+        acknowledgements = []
-        self.assertEqual(events, ["stopped"])
+        client_tools = SimpleNamespace(
-
+            on_interruption_processed=lambda: acknowledgements.append(True)
-    async def test_interrupted_output_rejects_stop_timeout(self):
-        async def wait_forever():
-            await asyncio.Event().wait()
-
-        with self.assertRaisesRegex(RuntimeError, "语音停止超时"):
-            await _wait_for_output_stop(
-                wait_forever,
-                timeout_seconds=0.001,
         )
 
+        bind_cascade_pipeline_events(
+            transport=transport,
+            worker=worker,
+            brain=brain,
+            context=SimpleNamespace(),
+            text_input=text_input,
+            user_aggregator=user_aggregator,
+            assistant_aggregator=assistant_aggregator,
+            greeting="",
+            vision_enabled=False,
+            vision_state={"client_id": None},
+            client_tools=client_tools,
+        )
+
+        await assistant_aggregator.handlers["on_interruption_processed"](
+            assistant_aggregator
+        )
+
+        self.assertEqual(acknowledgements, [True])
+
     async def test_greeting_keeps_playback_timestamp_until_client_ready(self):
         transport = _EventSource()
         text_input = _EventSource()