feat(azure-stt): add cancellation tracing and session guards

Author: oWretch

changelog/3884.changed.md

@@ -0,0 +1 @@
+- Added Azure STT cancellation tracing attributes and session termination guards so canceled recognition sessions surface structured observability data and stop accepting audio as if still healthy.

src/pipecat/services/azure/stt.py

@@ -31,7 +31,7 @@
 from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
-from pipecat.utils.tracing.service_decorators import traced_stt
+from pipecat.utils.tracing.service_decorators import trace_stt_cancellation, traced_stt
 
 try:
     from azure.cognitiveservices.speech import (
@@ -123,6 +123,10 @@ def __init__(
 
         self._audio_stream = None
         self._speech_recognizer = None
+        self._audio_sent = False
+        self._recognition_active = False
+        self._recognition_terminated = False
+        self._shutdown_requested = False
 
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate performance metrics.
@@ -179,7 +183,12 @@ async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
         try:
             await self.start_processing_metrics()
             if self._audio_stream:
+                if self._recognition_terminated and not self._shutdown_requested:
+                    logger.warning("Azure STT recognition terminated, dropping audio chunk")
+                    yield None
+                    return
                 self._audio_stream.write(audio)
+                self._audio_sent = True
             yield None
         except Exception as e:
             yield ErrorFrame(error=f"Unknown error occurred: {e}")
@@ -198,6 +207,11 @@ async def start(self, frame: StartFrame):
         if self._audio_stream:
             return
 
+        self._audio_sent = False
+        self._recognition_active = False
+        self._recognition_terminated = False
+        self._shutdown_requested = False
+
         try:
             stream_format = AudioStreamFormat(samples_per_second=self.sample_rate, channels=1)
             self._audio_stream = PushAudioInputStream(stream_format)
@@ -230,6 +244,10 @@ async def stop(self, frame: EndFrame):
         """
         await super().stop(frame)
 
+        self._shutdown_requested = True
+        self._recognition_active = False
+        self._recognition_terminated = True
+
         if self._speech_recognizer:
             self._speech_recognizer.stop_continuous_recognition_async()
 
@@ -246,6 +264,10 @@ async def cancel(self, frame: CancelFrame):
         """
         await super().cancel(frame)
 
+        self._shutdown_requested = True
+        self._recognition_active = False
+        self._recognition_terminated = True
+
         if self._speech_recognizer:
             self._speech_recognizer.stop_continuous_recognition_async()
 
@@ -259,6 +281,25 @@ async def _handle_transcription(
         """Handle a transcription result with tracing."""
         await self.stop_processing_metrics()
 
+    async def _trace_cancellation(
+        self,
+        *,
+        reason: str,
+        code: str,
+        recoverable: bool,
+        phase: str,
+    ):
+        """Record a trace span for a canceled Azure STT recognition."""
+        trace_stt_cancellation(
+            self,
+            error_type="azure.stt.canceled",
+            cancel_reason=reason,
+            cancel_code=code,
+            recoverable=recoverable,
+            phase=phase,
+            region=self._settings.region if isinstance(self._settings.region, str) else None,
+        )
+
     def _on_handle_recognized(self, event):
         if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0:
             language = getattr(event.result, "language", None) or self._settings.language
@@ -288,30 +329,87 @@ def _on_handle_recognizing(self, event):
 
     def _on_handle_canceled(self, event):
         details = getattr(event, "cancellation_details", None)
-        reason = getattr(details, "reason", "UNKNOWN")
-        code = getattr(details, "code", "UNKNOWN")
+        reason = self._normalize_cancellation_value(getattr(details, "reason", "UNKNOWN"))
+        code = self._normalize_cancellation_value(getattr(details, "code", "UNKNOWN"))
         error_details = getattr(details, "error_details", "")
+        phase = self._get_cancellation_phase()
+        recoverable = self._is_cancellation_recoverable(reason, code)
+
+        self._recognition_active = False
+        self._recognition_terminated = True
 
         logger.error(
-            "Azure STT recognition canceled: reason={}, code={}, details={}",
+            "Azure STT recognition canceled: reason={}, code={}, phase={}, recoverable={}, details={}",
             reason,
             code,
+            phase,
+            recoverable,
             error_details,
         )
 
-        error_message = f"Azure STT recognition canceled: {code} - {error_details}"
+        asyncio.run_coroutine_threadsafe(
+            self._trace_cancellation(
+                reason=reason,
+                code=code,
+                recoverable=recoverable,
+                phase=phase,
+            ),
+            self.get_event_loop(),
+        )
+
+        error_message = f"Azure STT recognition canceled: {reason} ({code})"
         asyncio.run_coroutine_threadsafe(
             self.push_error(error_msg=error_message), self.get_event_loop()
         )
 
     def _on_handle_session_started(self, event):
+        self._recognition_active = True
+        self._recognition_terminated = False
         logger.info(
             "Azure STT session started: session_id={}",
             getattr(event, "session_id", "unknown"),
         )
 
     def _on_handle_session_stopped(self, event):
-        logger.warning(
-            "Azure STT session stopped: session_id={}",
-            getattr(event, "session_id", "unknown"),
-        )
+        self._recognition_active = False
+        self._recognition_terminated = True
+        if self._shutdown_requested:
+            logger.info(
+                "Azure STT session stopped during shutdown: session_id={}",
+                getattr(event, "session_id", "unknown"),
+            )
+        else:
+            logger.warning(
+                "Azure STT session stopped: session_id={}",
+                getattr(event, "session_id", "unknown"),
+            )
+
+    @staticmethod
+    def _normalize_cancellation_value(value: Any) -> str:
+        normalized = getattr(value, "name", None)
+        if normalized:
+            return normalized
+        return str(value)
+
+    def _get_cancellation_phase(self) -> str:
+        if self._shutdown_requested:
+            return "shutdown"
+        if not self._recognition_active and not self._audio_sent:
+            return "startup"
+        return "streaming"
+
+    @staticmethod
+    def _is_cancellation_recoverable(reason: str, code: str) -> bool:
+        if reason == "CancelledByUser":
+            return True
+        if reason != "Error":
+            return False
+
+        return code in {
+            "ConnectionFailure",
+            "ServiceRedirectPermanent",
+            "ServiceRedirectTemporary",
+            "ServiceTimeout",
+            "ServiceUnavailable",
+            "TooManyRequests",
+        }

src/pipecat/utils/tracing/service_decorators.py

@@ -373,6 +373,51 @@ async def wrapper(self, transcript, is_final, language=None):
     return decorator
 
 
+def trace_stt_cancellation(
+    service,
+    *,
+    error_type: str,
+    cancel_reason: str,
+    cancel_code: str,
+    recoverable: bool,
+    phase: str,
+    region: Optional[str] = None,
+) -> None:
+    """Create a trace span for STT cancellation events.
+
+    Args:
+        service: STT service instance generating the cancellation.
+        error_type: Stable error classification.
+        cancel_reason: Provider cancellation reason.
+        cancel_code: Provider cancellation code.
+        recoverable: Whether the application should attempt recovery.
+        phase: Service lifecycle phase where cancellation happened.
+        region: Cloud region associated with the service, if known.
+    """
+    if not is_tracing_available() or not getattr(service, "_tracing_enabled", False):
+        return
+
+    service_class_name = service.__class__.__name__
+    parent_context = _get_turn_context(service) or _get_parent_service_context(service)
+
+    tracer = trace.get_tracer("pipecat")
+    with tracer.start_as_current_span("stt.cancel", context=parent_context) as current_span:
+        current_span.set_attribute(
+            "gen_ai.system", service_class_name.replace("STTService", "").lower()
+        )
+        current_span.set_attribute("gen_ai.operation.name", "stt.cancel")
+        current_span.set_attribute("error.type", error_type)
+        current_span.set_attribute("stt.cancel.reason", cancel_reason)
+        current_span.set_attribute("stt.cancel.code", cancel_code)
+        current_span.set_attribute("stt.cancel.recoverable", recoverable)
+        current_span.set_attribute("stt.cancel.phase", phase)
+        if region:
+            current_span.set_attribute("cloud.region", region)
+
+        if cancel_reason == "Error":
+            current_span.set_status(trace.Status(trace.StatusCode.ERROR, cancel_code))
+
+
 def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -> Callable:
     """Trace LLM service methods with LLM-specific attributes.
 

tests/test_azure_stt.py

@@ -39,12 +39,12 @@ class ConnectionClosedOK(Exception):
     class WebSocketClientProtocol:
         pass
 
-    protocol_module.State = State
-    exceptions_module.ConnectionClosedError = ConnectionClosedError
-    exceptions_module.ConnectionClosedOK = ConnectionClosedOK
-    websockets_module.protocol = protocol_module
-    websockets_module.exceptions = exceptions_module
-    websockets_module.WebSocketClientProtocol = WebSocketClientProtocol
+    setattr(protocol_module, "State", State)
+    setattr(exceptions_module, "ConnectionClosedError", ConnectionClosedError)
+    setattr(exceptions_module, "ConnectionClosedOK", ConnectionClosedOK)
+    setattr(websockets_module, "protocol", protocol_module)
+    setattr(websockets_module, "exceptions", exceptions_module)
+    setattr(websockets_module, "WebSocketClientProtocol", WebSocketClientProtocol)
 
 
 def _install_azure_speech_stub() -> None:
@@ -118,17 +118,17 @@ class AudioConfig:
         def __init__(self, stream=None):
             self.stream = stream
 
-    speech_module.ResultReason = ResultReason
-    speech_module.SpeechConfig = SpeechConfig
-    speech_module.SpeechRecognizer = SpeechRecognizer
-    audio_module.AudioStreamFormat = AudioStreamFormat
-    audio_module.PushAudioInputStream = PushAudioInputStream
-    dialog_module.AudioConfig = AudioConfig
+    setattr(speech_module, "ResultReason", ResultReason)
+    setattr(speech_module, "SpeechConfig", SpeechConfig)
+    setattr(speech_module, "SpeechRecognizer", SpeechRecognizer)
+    setattr(audio_module, "AudioStreamFormat", AudioStreamFormat)
+    setattr(audio_module, "PushAudioInputStream", PushAudioInputStream)
+    setattr(dialog_module, "AudioConfig", AudioConfig)
 
-    azure_module.cognitiveservices = cognitiveservices_module
-    cognitiveservices_module.speech = speech_module
-    speech_module.audio = audio_module
-    speech_module.dialog = dialog_module
+    setattr(azure_module, "cognitiveservices", cognitiveservices_module)
+    setattr(cognitiveservices_module, "speech", speech_module)
+    setattr(speech_module, "audio", audio_module)
+    setattr(speech_module, "dialog", dialog_module)
 
 
 _install_azure_speech_stub()
@@ -137,16 +137,16 @@ def __init__(self, stream=None):
 azure_package_module = types.ModuleType("pipecat.services.azure")
 azure_package_module.__path__ = []
 common_module = types.ModuleType("pipecat.services.azure.common")
-common_module.language_to_azure_language = lambda _language: "en-US"
+setattr(common_module, "language_to_azure_language", lambda _language: "en-US")
 
 sys.modules["pipecat.services.azure"] = azure_package_module
 sys.modules["pipecat.services.azure.common"] = common_module
 
 stt_file = pathlib.Path(__file__).resolve().parents[1] / "src/pipecat/services/azure/stt.py"
 spec = importlib.util.spec_from_file_location("pipecat.services.azure.stt", stt_file)
+assert spec is not None and spec.loader is not None
 stt_module = importlib.util.module_from_spec(spec)
 sys.modules["pipecat.services.azure.stt"] = stt_module
-assert spec and spec.loader
 spec.loader.exec_module(stt_module)
 
 from pipecat.frames.frames import StartFrame
@@ -192,6 +192,7 @@ def test_canceled_handler_pushes_error_with_details():
     service = stt_module.AzureSTTService(api_key="test-key", region="eastus")
     service.push_error = AsyncMock()
     service.get_event_loop = MagicMock(return_value=MagicMock())
+    service._trace_cancellation = AsyncMock()
 
     canceled_event = MagicMock()
     canceled_event.cancellation_details.reason = "Error"
@@ -202,12 +203,19 @@ def test_canceled_handler_pushes_error_with_details():
         service._on_handle_canceled(canceled_event)
 
     service.push_error.assert_called_once_with(
-        error_msg="Azure STT recognition canceled: AuthenticationFailure - 401 Unauthorized"
+        error_msg="Azure STT recognition canceled: Error (AuthenticationFailure)"
     )
-    assert run_threadsafe.call_count == 1
+    service._trace_cancellation.assert_called_once_with(
+        reason="Error",
+        code="AuthenticationFailure",
+        recoverable=False,
+        phase="startup",
+    )
+    assert run_threadsafe.call_count == 2
 
-    pending_coroutine = run_threadsafe.call_args.args[0]
-    pending_coroutine.close()
+    for call in run_threadsafe.call_args_list:
+        pending_coroutine = call.args[0]
+        pending_coroutine.close()
 
 
 def test_canceled_handler_uses_safe_defaults_when_details_missing():
@@ -216,6 +224,7 @@ def test_canceled_handler_uses_safe_defaults_when_details_missing():
     service = stt_module.AzureSTTService(api_key="test-key", region="eastus")
     service.push_error = AsyncMock()
     service.get_event_loop = MagicMock(return_value=MagicMock())
+    service._trace_cancellation = AsyncMock()
 
     canceled_event = MagicMock()
     canceled_event.cancellation_details = None
@@ -224,9 +233,33 @@ def test_canceled_handler_uses_safe_defaults_when_details_missing():
         service._on_handle_canceled(canceled_event)
 
     service.push_error.assert_called_once_with(
-        error_msg="Azure STT recognition canceled: UNKNOWN - "
+        error_msg="Azure STT recognition canceled: UNKNOWN (UNKNOWN)"
+    )
+    service._trace_cancellation.assert_called_once_with(
+        reason="UNKNOWN",
+        code="UNKNOWN",
+        recoverable=False,
+        phase="startup",
     )
-    assert run_threadsafe.call_count == 1
+    assert run_threadsafe.call_count == 2
+
+    for call in run_threadsafe.call_args_list:
+        pending_coroutine = call.args[0]
+        pending_coroutine.close()
+
+
+@pytest.mark.asyncio
+async def test_run_stt_drops_audio_after_terminated_session():
+    """Verify audio is not written after a terminated session."""
+
+    service = stt_module.AzureSTTService(api_key="test-key", region="eastus")
+    service.start_processing_metrics = AsyncMock()
+    service._audio_stream = MagicMock()
+    service._recognition_terminated = True
+
+    frames = []
+    async for frame in service.run_stt(b"audio"):
+        frames.append(frame)
 
-    pending_coroutine = run_threadsafe.call_args.args[0]
-    pending_coroutine.close()
+    service._audio_stream.write.assert_not_called()
+    assert frames == [None]