fix(realtime): complete workflow message transitions

backend/services/pipecat/pipeline.py

@@ -909,7 +909,22 @@ async def run_realtime_pipeline(
 
     call_end = CallEndCoordinator(queue_call_end)
     client_tools = ClientToolBroker()
-    client_tools.set_interrupt_handler(realtime.interrupt)
+
+    async def interrupt_for_client_tool_result() -> None:
+        """Release confirmations after the realtime response is interrupted.
+
+        Fixed Message speech intentionally suppresses the provider transcript,
+        so the text aggregator may have no assistant turn from which to emit an
+        ``on_interruption_processed`` event.  The realtime adapter owns and
+        clears that response state itself.  Wait for the provider boundary as
+        well, matching Pipeline's rule that the next node cannot generate until
+        the interrupted response has fully finished.
+        """
+        await realtime.interrupt()
+        await realtime.wait_for_response_boundary()
+        client_tools.on_interruption_processed()
+
+    client_tools.set_interrupt_handler(interrupt_for_client_tool_result)
     user_input = RealtimeUserInputProcessor(
         should_ignore_input=lambda: (
             call_end.ending or not input_state["enabled"]

backend/services/pipecat/qwen_audio_realtime.py

@@ -27,6 +27,7 @@ from pipecat.frames.frames import (
     OutputTransportMessageUrgentFrame,
     StartFrame,
     TTSAudioRawFrame,
+    TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
@@ -120,6 +121,7 @@ class QwenAudioRealtimeService(AIService):
         self._assistant_text = ""
         self._assistant_timestamp = ""
         self._greeting_request_item_id: str | None = None
+        self._entry_trigger_item_id: str | None = None
         self._user_transcript_pending = False
         self._user_transcript_item_id = ""
         self._user_transcript_timestamp = ""
@@ -215,9 +217,27 @@ class QwenAudioRealtimeService(AIService):
         await self._cancel_active_response()
         await self.broadcast_interruption()
 
-    async def request_response(self) -> None:
+    async def request_response(self, *, trigger_text: str | None = None) -> None:
+        if trigger_text:
+            item_id = f"item_{uuid4().hex}"
+            self._entry_trigger_item_id = item_id
+            await self._send_event(
+                {
+                    "type": "conversation.item.create",
+                    "item": {
+                        "id": item_id,
+                        "type": "message",
+                        "role": "user",
+                        "content": [{"type": "input_text", "text": trigger_text}],
+                    },
+                }
+            )
         await self._send_event({"type": "response.create"})
 
+    async def wait_for_response_boundary(self) -> None:
+        """Wait until Qwen has finished or cancelled the current response."""
+        await self._response_done.wait()
+
     def set_speech_started_handler(
         self,
         handler: SpeechStartedHandler | None,
@@ -387,6 +407,7 @@ class QwenAudioRealtimeService(AIService):
         self._deferred_assistant_messages.clear()
         self._tool_session.clear()
         self._function_names.clear()
+        self._entry_trigger_item_id = None
         self._resolve_fixed_speech()
         if websocket and websocket.state is State.OPEN:
             try:
@@ -412,6 +433,7 @@ class QwenAudioRealtimeService(AIService):
                 self._session_ready.clear()
             self._response_done.set()
             self._greeting_request_item_id = None
+            self._entry_trigger_item_id = None
             self._resolve_fixed_speech()
             if self._receive_task is asyncio.current_task():
                 self._receive_task = None
@@ -480,8 +502,10 @@ class QwenAudioRealtimeService(AIService):
             status = response.get("status") if isinstance(response, dict) else None
             interrupted = status in {"cancelled", "incomplete", "interrupted", "failed"}
             self._response_active = False
+            await self.push_frame(TTSStoppedFrame())
             await self._finish_assistant_text(interrupted=interrupted)
             await self._delete_greeting_request()
+            await self._delete_entry_trigger()
             self._resolve_fixed_speech()
             self._response_done.set()
         elif event_type == "response.output_item.added":
@@ -579,6 +603,15 @@ class QwenAudioRealtimeService(AIService):
                 wait_until_ready=False,
             )
 
+    async def _delete_entry_trigger(self) -> None:
+        item_id = self._entry_trigger_item_id
+        self._entry_trigger_item_id = None
+        if item_id:
+            await self._send_event(
+                {"type": "conversation.item.delete", "item_id": item_id},
+                wait_until_ready=False,
+            )
+
     async def _start_user_transcript_turn(
         self,
         event: dict[str, Any],
@@ -643,6 +676,10 @@ class QwenAudioRealtimeService(AIService):
     async def _send_event(
         self, payload: dict[str, Any], *, wait_until_ready: bool = True
     ) -> None:
+        if payload.get("type") == "response.create":
+            # Mark the response busy before the server's response.created
+            # arrives so an immediate confirmation cannot pass the boundary.
+            self._response_done.clear()
         if wait_until_ready and not self._session_ready.is_set():
             self._pending_events.append(payload)
             return

backend/services/pipecat/stepfun_realtime.py

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
     OutputTransportMessageUrgentFrame,
     StartFrame,
     TTSAudioRawFrame,
+    TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
@@ -85,6 +86,7 @@ class StepFunRealtimeService(AIService):
         )
         self._fixed_speech_completion: asyncio.Future[None] | None = None
         self._fixed_speech_request_item_id: str | None = None
+        self._entry_trigger_item_id: str | None = None
         self._suppress_response_transcript = False
         self._speech_started_handler: SpeechStartedHandler | None = None
         self._function_names: dict[str, str] = {}
@@ -164,9 +166,27 @@ class StepFunRealtimeService(AIService):
         self._resolve_fixed_speech()
         await self.broadcast_interruption()
 
-    async def request_response(self) -> None:
+    async def request_response(self, *, trigger_text: str | None = None) -> None:
+        if trigger_text:
+            item_id = f"item_{uuid4().hex}"
+            self._entry_trigger_item_id = item_id
+            await self._send_event(
+                {
+                    "type": "conversation.item.create",
+                    "item": {
+                        "id": item_id,
+                        "type": "message",
+                        "role": "user",
+                        "content": [{"type": "input_text", "text": trigger_text}],
+                    },
+                }
+            )
         await self._send_event({"type": "response.create"})
 
+    async def wait_for_response_boundary(self) -> None:
+        """Wait until StepFun has finished or cancelled the current response."""
+        await self._response_done.wait()
+
     def set_speech_started_handler(
         self,
         handler: SpeechStartedHandler | None,
@@ -246,6 +266,7 @@ class StepFunRealtimeService(AIService):
         self._tool_session.clear()
         self._function_names.clear()
         self._fixed_speech_request_item_id = None
+        self._entry_trigger_item_id = None
         self._resolve_fixed_speech()
         if websocket and websocket.state is State.OPEN:
             try:
@@ -280,6 +301,7 @@ class StepFunRealtimeService(AIService):
                 self._session_ready.clear()
             self._response_done.set()
             self._fixed_speech_request_item_id = None
+            self._entry_trigger_item_id = None
             self._resolve_fixed_speech()
             if self._receive_task is asyncio.current_task():
                 self._receive_task = None
@@ -331,8 +353,10 @@ class StepFunRealtimeService(AIService):
                 "incomplete",
                 "interrupted",
             }
+            await self.push_frame(TTSStoppedFrame())
             await self._finish_assistant_text(interrupted=interrupted)
             await self._delete_fixed_speech_request()
+            await self._delete_entry_trigger()
             self._resolve_fixed_speech()
             self._response_done.set()
         elif event_type == "response.output_item.added":
@@ -474,9 +498,22 @@ class StepFunRealtimeService(AIService):
                 wait_until_ready=False,
             )
 
+    async def _delete_entry_trigger(self) -> None:
+        item_id = self._entry_trigger_item_id
+        self._entry_trigger_item_id = None
+        if item_id:
+            await self._send_event(
+                {"type": "conversation.item.delete", "item_id": item_id},
+                wait_until_ready=False,
+            )
+
     async def _send_event(
         self, payload: dict[str, Any], *, wait_until_ready: bool = True
     ) -> None:
+        if payload.get("type") == "response.create":
+            # Mark the response busy before the server's response.created
+            # arrives so an immediate confirmation cannot pass the boundary.
+            self._response_done.clear()
         if wait_until_ready and not self._session_ready.is_set():
             self._pending_events.append(payload)
             return

backend/services/workflow/realtime.py

@@ -35,6 +35,11 @@ from services.workflow_engine import WorkflowEngine
 
 
 MAX_AUTOMATIC_HOPS = 50
+AGENT_ENTRY_TRIGGER = (
+    "[工作流内部事件，不是用户消息] 当前 Agent 节点已激活。"
+    "请严格依据当前 instructions 立即生成进入该节点时的回复，"
+    "不要提及或复述本事件。"
+)
 ToolHandler = Callable[[dict[str, Any]], Awaitable[RealtimeToolResult]]
 
 
@@ -80,10 +85,18 @@ class RealtimeWorkflowOutput(WorkflowOutput):
                 **({"nodeId": node_id} if node_id else {}),
             }
         )
-        return await self._realtime.speak_fixed(
+        track_speech = getattr(self._runtime.call_end, "track_speech", None)
+        playback_completion: Awaitable[None] | None = None
+        if callable(track_speech):
+            playback_completion = track_speech()
+        provider_completion = await self._realtime.speak_fixed(
             content,
             suppress_transcript=True,
         )
+        # Message playback policy and deterministic continuation must use the
+        # transport boundary.  Provider response.done only means generation
+        # has finished; audio may still be buffered at the output transport.
+        return playback_completion or provider_completion
 
 
 class WorkflowRealtimeController:
@@ -138,7 +151,7 @@ class WorkflowRealtimeController:
             allow_visible_actions=True,
         )
         if activation.continue_response:
-            await self._runtime.realtime.request_response()
+            await self._request_agent_entry_response()
 
     async def on_client_ready(self) -> None:
         await self._output.mark_client_ready()
@@ -160,7 +173,7 @@ class WorkflowRealtimeController:
         if activation and activation.after_output:
             await activation.after_output()
         elif activation and activation.continue_response:
-            await self._runtime.realtime.request_response()
+            await self._request_agent_entry_response()
         return SessionVariableUpdate(
             changed=changed,
             dynamic_variables=self._store.public_values(),
@@ -321,7 +334,13 @@ class WorkflowRealtimeController:
         if activation.after_output:
             await activation.after_output()
         elif activation.continue_response:
-            await self._runtime.realtime.request_response()
+            await self._request_agent_entry_response()
+
+    async def _request_agent_entry_response(self) -> None:
+        """Give provider inference a temporary turn for Agent entry speech."""
+        await self._runtime.realtime.request_response(
+            trigger_text=AGENT_ENTRY_TRIGGER,
+        )
 
     async def _follow_edge(
         self,