feat: enhance call end coordination with state management

- Added support for tracking call end states (pending, speaking, cancelled) in CallEndCoordinator.
- Introduced a callback mechanism for state changes to allow external handling of call end states.
- Updated pipeline and tests to accommodate new state management features, ensuring proper handling of call end scenarios.

backend/services/pipecat/pipeline.py

@@ -201,6 +201,23 @@ def _multimodal_user_input_frame(
     )
 
 
+def _call_end_state_frame(
+    state: str,
+    *,
+    interruptible: bool,
+    voice_interruptible: bool,
+) -> OutputTransportMessageUrgentFrame:
+    """Build the client event used to render the pending call-end state."""
+    return OutputTransportMessageUrgentFrame(
+        message={
+            "type": "call-end-state",
+            "state": state,
+            "interruptible": interruptible,
+            "voice_interruptible": voice_interruptible,
+        }
+    )
+
+
 async def run_pipeline(
     transport,
     cfg: AssistantConfig,
@@ -295,7 +312,24 @@ async def run_pipeline(
         )
         await worker.queue_frame(EndFrame())
 
-    call_end = CallEndCoordinator(queue_call_end)
+    async def queue_call_end_state(state: str, interruptible: bool) -> None:
+        worker = worker_holder.get("worker")
+        if worker is None:
+            return
+        await worker.queue_frame(
+            _call_end_state_frame(
+                state,
+                interruptible=interruptible,
+                voice_interruptible=(
+                    interruptible and current_enable_interrupt
+                ),
+            )
+        )
+
+    call_end = CallEndCoordinator(
+        queue_call_end,
+        on_state_change=queue_call_end_state,
+    )
 
     knowledge_config = cfg.knowledge_retrieval_config
     knowledge_mode = str(knowledge_config.get("mode", "automatic"))
@@ -976,7 +1010,24 @@ async def run_realtime_pipeline(
         )
         await worker.queue_frame(EndFrame())
 
-    call_end = CallEndCoordinator(queue_call_end)
+    async def queue_call_end_state(state: str, interruptible: bool) -> None:
+        worker = worker_holder.get("worker")
+        if worker is None:
+            return
+        await worker.queue_frame(
+            _call_end_state_frame(
+                state,
+                interruptible=interruptible,
+                voice_interruptible=(
+                    interruptible and cfg.enableInterrupt
+                ),
+            )
+        )
+
+    call_end = CallEndCoordinator(
+        queue_call_end,
+        on_state_change=queue_call_end_state,
+    )
     client_tools = ClientToolBroker()
 
     async def interrupt_for_client_tool_result() -> None: