codex fixex

src/agent/runtime/assistant.py

@@ -86,16 +86,12 @@ class Assistant(Agent):
         """Called when the agent enters the session. Set up metrics listeners."""
 
         def metrics_wrapper(event: MetricsCollectedEvent) -> None:
-            asyncio.create_task(
-                self._metrics_collector.on_metrics_collected(event.metrics)
-            )
+            self._metrics_collector.submit_metrics_collected(event.metrics)
 
         def transcript_wrapper(event: UserInputTranscribedEvent) -> None:
-            asyncio.create_task(
-                self._metrics_collector.on_user_input_transcribed(
-                    event.transcript,
-                    is_final=event.is_final,
-                )
+            self._metrics_collector.submit_user_input_transcribed(
+                event.transcript,
+                is_final=event.is_final,
             )
 
         def conversation_item_wrapper(event: ConversationItemAddedEvent) -> None:
@@ -103,27 +99,21 @@ class Assistant(Agent):
             role = getattr(item, "role", None)
             content = getattr(item, "content", None)
             item_created_at = getattr(item, "created_at", None)
-            asyncio.create_task(
-                self._metrics_collector.on_conversation_item_added(
-                    role=role,
-                    content=content,
-                    event_created_at=event.created_at,
-                    item_created_at=item_created_at,
-                )
+            self._metrics_collector.submit_conversation_item_added(
+                role=role,
+                content=content,
+                event_created_at=event.created_at,
+                item_created_at=item_created_at,
             )
 
         def speech_created_wrapper(event: SpeechCreatedEvent) -> None:
-            asyncio.create_task(
-                self._metrics_collector.on_speech_created(event.speech_handle)
-            )
+            self._metrics_collector.submit_speech_created(event.speech_handle)
 
         def function_tools_executed_wrapper(event: FunctionToolsExecutedEvent) -> None:
-            asyncio.create_task(
-                self._metrics_collector.on_function_tools_executed(
-                    function_calls=event.function_calls,
-                    function_call_outputs=event.function_call_outputs,
-                    created_at=event.created_at,
-                )
+            self._metrics_collector.submit_function_tools_executed(
+                function_calls=event.function_calls,
+                function_call_outputs=event.function_call_outputs,
+                created_at=event.created_at,
             )
             if self._tool_feedback is not None:
                 asyncio.create_task(
@@ -131,11 +121,9 @@ class Assistant(Agent):
                 )
 
         def agent_state_changed_wrapper(event: AgentStateChangedEvent) -> None:
-            asyncio.create_task(
-                self._metrics_collector.on_agent_state_changed(
-                    old_state=event.old_state,
-                    new_state=event.new_state,
-                )
+            self._metrics_collector.submit_agent_state_changed(
+                old_state=event.old_state,
+                new_state=event.new_state,
             )
 
         def error_wrapper(event: ErrorEvent) -> None:

src/agent/runtime/session.py

@@ -109,16 +109,6 @@ async def session_handler(ctx: agents.JobContext) -> None:
     startup_greeting_task: asyncio.Task[Any] | None = None
     tool_feedback = ToolFeedbackController(enabled=False)
 
-    if trace_provider:
-
-        async def flush_trace(_: str) -> None:
-            try:
-                trace_provider.force_flush()
-            except Exception as exc:
-                logger.warning(f"Failed to flush Langfuse traces: {exc}")
-
-        ctx.add_shutdown_callback(flush_trace)
-
     async def cancel_startup_greeting(_: str) -> None:
         await cancel_task_for_shutdown(
             startup_greeting_task,
@@ -147,6 +137,22 @@ async def session_handler(ctx: agents.JobContext) -> None:
         langfuse_enabled=trace_provider is not None,
     )
 
+    async def drain_pending_traces(_: str) -> None:
+        try:
+            await metrics_collector.drain_pending_traces()
+        except TimeoutError:
+            logger.warning("Timed out while draining pending Langfuse traces during shutdown")
+        except Exception as exc:
+            logger.warning(f"Failed to drain pending Langfuse traces: {exc}")
+        if trace_provider is None:
+            return
+        try:
+            trace_provider.force_flush()
+        except Exception as exc:
+            logger.warning(f"Failed to flush Langfuse traces: {exc}")
+
+    ctx.add_shutdown_callback(drain_pending_traces)
+
     if isinstance(ctx.job.metadata, str) and ctx.job.metadata.strip():
         try:
             metadata = json.loads(ctx.job.metadata)
@@ -158,11 +164,9 @@ async def session_handler(ctx: agents.JobContext) -> None:
             metadata.get("participant_id"),
             ctx.room.name,
         )
-        asyncio.create_task(
-            metrics_collector.on_session_metadata(
-                session_id=metadata.get("session_id"),
-                participant_id=metadata.get("participant_id"),
-            )
+        await metrics_collector.on_session_metadata(
+            session_id=metadata.get("session_id"),
+            participant_id=metadata.get("participant_id"),
         )
 
     tts_engine = create_tts()

src/agent/traces/metrics_collector.py

@@ -12,7 +12,7 @@ import uuid
 from collections import deque
 from dataclasses import asdict, dataclass
 from time import monotonic, time
-from typing import Any, Optional, Sequence, Union
+from typing import Any, Awaitable, Callable, Optional, Sequence, Union
 
 from livekit import rtc
 from livekit.agents import metrics
@@ -213,6 +213,16 @@ class PendingUserUtterance:
     watchdog_id: Optional[str] = None
 
 
+@dataclass
+class QueuedCollectorEvent:
+    """FIFO collector event that must be processed in-order."""
+
+    handler: Callable[..., Awaitable[Any]]
+    args: tuple[Any, ...]
+    kwargs: dict[str, Any]
+    waiter: asyncio.Future[Any] | None = None
+
+
 # ------------------------------------------------------------------
 # Facade
 # ------------------------------------------------------------------
@@ -261,6 +271,9 @@ class MetricsCollector:
         self._latest_vad_metrics: Optional[VADMetrics] = None
         self._latest_vad_metric_attributes: Optional[dict[str, Any]] = None
         self._first_final_user_turn_logged = False
+        self._event_queue: deque[QueuedCollectorEvent] = deque()
+        self._event_worker_task: asyncio.Task[None] | None = None
+        self._event_loop: asyncio.AbstractEventLoop | None = None
         self._llm_stall_timeout_sec = max(
             float(
                 getattr(
@@ -271,6 +284,19 @@ class MetricsCollector:
             ),
             0.0,
         )
+        self._shutdown_drain_timeout_sec = (
+            max(
+                float(
+                    getattr(
+                        settings.langfuse,
+                        "LANGFUSE_SHUTDOWN_DRAIN_TIMEOUT_MS",
+                        3000.0,
+                    )
+                ),
+                0.0,
+            )
+            / 1000.0
+        )
 
         self._tracer = TurnTracer(
             publisher=self._publisher,
@@ -300,6 +326,61 @@ class MetricsCollector:
     def _trace_post_tool_response_timeout_sec(self, value: float) -> None:
         self._tracer._trace_post_tool_response_timeout_sec = value
 
+    def submit_metrics_collected(self, collected_metrics: Any) -> None:
+        self._submit_serialized(self._handle_metrics_collected, collected_metrics)
+
+    def submit_user_input_transcribed(self, transcript: str, *, is_final: bool) -> None:
+        self._submit_serialized(
+            self._handle_user_input_transcribed,
+            transcript,
+            is_final=is_final,
+        )
+
+    def submit_conversation_item_added(
+        self,
+        *,
+        role: Optional[str],
+        content: Any,
+        event_created_at: Optional[float] = None,
+        item_created_at: Optional[float] = None,
+    ) -> None:
+        self._submit_serialized(
+            self._handle_conversation_item_added,
+            role=role,
+            content=content,
+            event_created_at=event_created_at,
+            item_created_at=item_created_at,
+        )
+
+    def submit_speech_created(self, speech_handle: Any) -> None:
+        self._submit_serialized(self._handle_speech_created, speech_handle)
+
+    def submit_function_tools_executed(
+        self,
+        *,
+        function_calls: list[Any],
+        function_call_outputs: list[Any],
+        created_at: float,
+    ) -> None:
+        self._submit_serialized(
+            self._handle_function_tools_executed,
+            function_calls=function_calls,
+            function_call_outputs=function_call_outputs,
+            created_at=created_at,
+        )
+
+    def submit_agent_state_changed(
+        self,
+        *,
+        old_state: str,
+        new_state: str,
+    ) -> None:
+        self._submit_serialized(
+            self._handle_agent_state_changed,
+            old_state=old_state,
+            new_state=new_state,
+        )
+
     # ------------------------------------------------------------------
     # Public event handlers
     # ------------------------------------------------------------------
@@ -309,6 +390,18 @@ class MetricsCollector:
         *,
         session_id: Any,
         participant_id: Any,
+    ) -> None:
+        await self._call_serialized(
+            self._handle_session_metadata,
+            session_id=session_id,
+            participant_id=participant_id,
+        )
+
+    async def _handle_session_metadata(
+        self,
+        *,
+        session_id: Any,
+        participant_id: Any,
     ) -> None:
         normalized_session = _normalize(session_id)
         normalized_participant = _normalize(participant_id)
@@ -322,6 +415,18 @@ class MetricsCollector:
         transcript: str,
         *,
         is_final: bool,
+    ) -> None:
+        await self._call_serialized(
+            self._handle_user_input_transcribed,
+            transcript,
+            is_final=is_final,
+        )
+
+    async def _handle_user_input_transcribed(
+        self,
+        transcript: str,
+        *,
+        is_final: bool,
     ) -> None:
         if not is_final:
             return
@@ -364,6 +469,22 @@ class MetricsCollector:
         content: Any,
         event_created_at: Optional[float] = None,
         item_created_at: Optional[float] = None,
+    ) -> None:
+        await self._call_serialized(
+            self._handle_conversation_item_added,
+            role=role,
+            content=content,
+            event_created_at=event_created_at,
+            item_created_at=item_created_at,
+        )
+
+    async def _handle_conversation_item_added(
+        self,
+        *,
+        role: Optional[str],
+        content: Any,
+        event_created_at: Optional[float] = None,
+        item_created_at: Optional[float] = None,
     ) -> None:
         if role not in {"user", "assistant"}:
             return
@@ -408,6 +529,20 @@ class MetricsCollector:
         function_calls: list[Any],
         function_call_outputs: list[Any],
         created_at: float,
+    ) -> None:
+        await self._call_serialized(
+            self._handle_function_tools_executed,
+            function_calls=function_calls,
+            function_call_outputs=function_call_outputs,
+            created_at=created_at,
+        )
+
+    async def _handle_function_tools_executed(
+        self,
+        *,
+        function_calls: list[Any],
+        function_call_outputs: list[Any],
+        created_at: float,
     ) -> None:
         trace_turn = await self._tracer.attach_function_tools_executed(
             function_calls=function_calls,
@@ -418,12 +553,18 @@ class MetricsCollector:
         await self._tracer.maybe_finalize(trace_turn)
 
     async def on_tool_step_started(self) -> bool:
+        return await self._call_serialized(self._handle_tool_step_started)
+
+    async def _handle_tool_step_started(self) -> bool:
         trace_turn, should_announce = await self._tracer.attach_tool_step_started()
         await self._publish_partial_turn_pipeline_summary(trace_turn)
         await self._tracer.maybe_finalize(trace_turn)
         return should_announce
 
     async def on_speech_created(self, speech_handle: Any) -> None:
+        await self._call_serialized(self._handle_speech_created, speech_handle)
+
+    async def _handle_speech_created(self, speech_handle: Any) -> None:
         speech_id = _normalize(getattr(speech_handle, "id", None))
         if speech_id:
             self._pending_speech_ids_for_first_audio.append(speech_id)
@@ -435,6 +576,7 @@ class MetricsCollector:
             await self._on_assistant_text(
                 assistant_text,
                 event_created_at=assistant_created_at,
+                speech_id=speech_id,
             )
 
         add_done_callback = getattr(speech_handle, "add_done_callback", None)
@@ -444,18 +586,15 @@ class MetricsCollector:
         def _on_done(handle: Any) -> None:
             try:
                 done_speech_id = _normalize(getattr(handle, "id", None))
-                if done_speech_id:
-                    self._discard_pending_speech_id(done_speech_id)
                 text, created_at = _extract_latest_assistant_chat_item(
                     getattr(handle, "chat_items", [])
                 )
-                if text:
-                    asyncio.create_task(
-                        self._on_assistant_text(
-                            text,
-                            event_created_at=created_at,
-                        )
-                    )
+                self._submit_serialized(
+                    self._handle_speech_done,
+                    done_speech_id,
+                    text,
+                    created_at,
+                )
             except Exception:
                 return
 
@@ -464,11 +603,38 @@ class MetricsCollector:
         except Exception:
             return
 
+    async def _handle_speech_done(
+        self,
+        speech_id: Optional[str],
+        assistant_text: str,
+        event_created_at: Optional[float],
+    ) -> None:
+        if speech_id:
+            self._discard_pending_speech_id(speech_id)
+        if assistant_text:
+            await self._on_assistant_text(
+                assistant_text,
+                event_created_at=event_created_at,
+                speech_id=speech_id,
+            )
+
     async def on_agent_state_changed(
         self,
         *,
         old_state: str,
         new_state: str,
+    ) -> None:
+        await self._call_serialized(
+            self._handle_agent_state_changed,
+            old_state=old_state,
+            new_state=new_state,
+        )
+
+    async def _handle_agent_state_changed(
+        self,
+        *,
+        old_state: str,
+        new_state: str,
     ) -> None:
         if new_state != "speaking":
             return
@@ -498,6 +664,10 @@ class MetricsCollector:
                 old_state,
                 new_state,
             )
+            await self._tracer.mark_first_audio_started(
+                speech_id=speech_id,
+                started_at=time(),
+            )
 
     async def on_tts_synthesized(
         self,
@@ -505,6 +675,20 @@ class MetricsCollector:
         ttfb: float,
         duration: float,
         audio_duration: float,
+    ) -> None:
+        await self._call_serialized(
+            self._handle_tts_synthesized,
+            ttfb=ttfb,
+            duration=duration,
+            audio_duration=audio_duration,
+        )
+
+    async def _handle_tts_synthesized(
+        self,
+        *,
+        ttfb: float,
+        duration: float,
+        audio_duration: float,
     ) -> None:
         if ttfb < 0:
             return
@@ -558,6 +742,21 @@ class MetricsCollector:
             metrics.EOUMetrics,
             metrics.VADMetrics,
         ],
+    ) -> None:
+        await self._call_serialized(
+            self._handle_metrics_collected,
+            collected_metrics,
+        )
+
+    async def _handle_metrics_collected(
+        self,
+        collected_metrics: Union[
+            metrics.STTMetrics,
+            metrics.LLMMetrics,
+            metrics.TTSMetrics,
+            metrics.EOUMetrics,
+            metrics.VADMetrics,
+        ],
     ) -> None:
         speech_id = None
         turn_metrics = None
@@ -595,8 +794,6 @@ class MetricsCollector:
             speech_id = collected_metrics.speech_id or collected_metrics.request_id
             turn_metrics = self._get_or_create_turn(speech_id, role="agent")
             self._latest_agent_speech_id = speech_id
-            if self._pending_agent_transcripts and not turn_metrics.transcript:
-                turn_metrics.transcript = self._pending_agent_transcripts.popleft()
             turn_metrics.llm = LLMMetrics(
                 type=collected_metrics.type,
                 label=collected_metrics.label,
@@ -727,12 +924,98 @@ class MetricsCollector:
         await self._tracer.maybe_finalize(trace_turn)
 
     async def wait_for_pending_trace_tasks(self) -> None:
+        await self._wait_for_event_queue_idle()
         await self._tracer.wait_for_pending_tasks()
 
+    async def drain_pending_traces(self) -> None:
+        if self._shutdown_drain_timeout_sec <= 0.0:
+            return
+        await asyncio.wait_for(
+            self._drain_pending_traces_once(),
+            timeout=self._shutdown_drain_timeout_sec,
+        )
+
     # ------------------------------------------------------------------
     # Internal helpers
     # ------------------------------------------------------------------
 
+    async def _drain_pending_traces_once(self) -> None:
+        await self._wait_for_event_queue_idle()
+        await self._tracer.drain_pending_turns()
+        await self._tracer.wait_for_pending_tasks()
+
+    async def _call_serialized(
+        self,
+        handler: Callable[..., Awaitable[Any]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> Any:
+        loop = asyncio.get_running_loop()
+        waiter: asyncio.Future[Any] = loop.create_future()
+        self._enqueue_serialized(handler, args=args, kwargs=kwargs, waiter=waiter)
+        return await waiter
+
+    def _submit_serialized(
+        self,
+        handler: Callable[..., Awaitable[Any]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        self._enqueue_serialized(handler, args=args, kwargs=kwargs, waiter=None)
+
+    def _enqueue_serialized(
+        self,
+        handler: Callable[..., Awaitable[Any]],
+        *,
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+        waiter: asyncio.Future[Any] | None,
+    ) -> None:
+        loop = asyncio.get_running_loop()
+        if self._event_loop is None:
+            self._event_loop = loop
+        elif self._event_loop is not loop:
+            raise RuntimeError("MetricsCollector cannot be shared across event loops")
+
+        self._event_queue.append(
+            QueuedCollectorEvent(
+                handler=handler,
+                args=args,
+                kwargs=kwargs,
+                waiter=waiter,
+            )
+        )
+        if self._event_worker_task is None:
+            self._event_worker_task = loop.create_task(self._run_event_worker())
+
+    async def _run_event_worker(self) -> None:
+        while True:
+            if not self._event_queue:
+                self._event_worker_task = None
+                return
+
+            event = self._event_queue.popleft()
+            try:
+                result = await event.handler(*event.args, **event.kwargs)
+            except Exception as exc:
+                if event.waiter is not None and not event.waiter.done():
+                    event.waiter.set_exception(exc)
+                else:
+                    logger.exception(
+                        "Metrics collector event processing failed: handler=%s",
+                        getattr(event.handler, "__name__", repr(event.handler)),
+                    )
+            else:
+                if event.waiter is not None and not event.waiter.done():
+                    event.waiter.set_result(result)
+
+    async def _wait_for_event_queue_idle(self) -> None:
+        while self._event_worker_task is not None:
+            task = self._event_worker_task
+            await asyncio.gather(task, return_exceptions=True)
+            if self._event_worker_task is task:
+                break
+
     def _get_or_create_state(self, speech_id: str) -> TurnState:
         if speech_id not in self._turns:
             self._turns[speech_id] = TurnState()
@@ -753,14 +1036,15 @@ class MetricsCollector:
         assistant_text: str,
         *,
         event_created_at: Optional[float] = None,
+        speech_id: Optional[str] = None,
     ) -> None:
         normalized = assistant_text.strip()
         if not normalized:
             return
-        _append_if_new(self._pending_agent_transcripts, normalized)
         trace_turn = await self._tracer.attach_assistant_text(
             normalized,
             event_created_at=event_created_at,
+            speech_id=speech_id,
         )
         await self._tracer.maybe_finalize(trace_turn)
 
@@ -840,7 +1124,7 @@ class MetricsCollector:
 
     def _current_open_user_utterance(self) -> Optional[PendingUserUtterance]:
         utterance = self._latest_user_utterance()
-        if utterance is None or utterance.committed:
+        if utterance is None or utterance.committed or utterance.llm_started:
             return None
         return utterance
 

src/agent/traces/turn_tracer.py

@@ -76,6 +76,11 @@ class TraceTurn:
     tts_updated_order: Optional[int] = None
     event_counter: int = 0
     tool_post_response_missing: bool = False
+    assistant_audio_started: bool = False
+    assistant_audio_started_at: Optional[float] = None
+    interrupted: bool = False
+    interrupted_reason: Optional[str] = None
+    orphan_assistant_cutoff_at: Optional[float] = None
     coalesced_turn_ids: list[str] = field(default_factory=list)
     coalesced_user_transcripts: list[str] = field(default_factory=list)
     coalesced_fragment_count: int = 0
@@ -130,6 +135,14 @@ class TTSCallTrace:
     first_audio_at: Optional[float] = None
 
 
+@dataclass
+class AssistantTextRecord:
+    """Buffered assistant text that has not been correlated safely yet."""
+
+    text: str
+    event_created_at: Optional[float] = None
+
+
 @dataclass
 class TimelineEvent:
     """Ordered event for building trace phases."""
@@ -193,6 +206,10 @@ class TurnTracer:
         self._pending_agent_transcripts = pending_agent_transcripts
 
         self._pending_trace_turns: deque[TraceTurn] = deque()
+        self._pending_agent_transcripts_by_speech_id: dict[
+            str, deque[AssistantTextRecord]
+        ] = {}
+        self._orphan_assistant_text_records: deque[AssistantTextRecord] = deque()
         self._trace_lock = asyncio.Lock()
         self._trace_emit_tasks: set[asyncio.Task[None]] = set()
         self._trace_finalize_tasks: dict[str, asyncio.Task[None]] = {}
@@ -305,33 +322,38 @@ class TurnTracer:
     # ------------------------------------------------------------------
 
     async def create_turn(self, *, user_transcript: str, room_id: str) -> None:
+        completed_turns: list[TraceTurn] = []
         async with self._trace_lock:
             normalized = user_transcript.strip()
             if not normalized:
                 return
 
+            completed_turns = self._finalize_interrupted_turns_before_new_user_turn_locked()
+
             current_turn = self._latest_turn_where(lambda c: not c.user_turn_committed)
             if current_turn is not None:
                 self._update_user_turn_text(current_turn, normalized)
-                return
+            else:
+                new_turn = TraceTurn(
+                    turn_id=str(uuid.uuid4()),
+                    session_id=self._session_id,
+                    room_id=room_id,
+                    participant_id=self._participant_id,
+                    user_transcript=normalized,
+                    prompt_text=normalized,
+                )
+                new_turn.user_transcript_updated_at = new_turn.created_at
 
-            new_turn = TraceTurn(
-                turn_id=str(uuid.uuid4()),
-                session_id=self._session_id,
-                room_id=room_id,
-                participant_id=self._participant_id,
-                user_transcript=normalized,
-                prompt_text=normalized,
-            )
-            new_turn.user_transcript_updated_at = new_turn.created_at
+                coalesced_turn = self._coalesced_turn_candidate()
+                if coalesced_turn is not None:
+                    self._absorb_coalesced_turn_metadata(new_turn, coalesced_turn)
+                    self._pending_trace_turns.remove(coalesced_turn)
+                    self._cancel_finalize_timeout(coalesced_turn.turn_id)
 
-            coalesced_turn = self._coalesced_turn_candidate()
-            if coalesced_turn is not None:
-                self._absorb_coalesced_turn_metadata(new_turn, coalesced_turn)
-                self._pending_trace_turns.remove(coalesced_turn)
-                self._cancel_finalize_timeout(coalesced_turn.turn_id)
+                self._pending_trace_turns.append(new_turn)
 
-            self._pending_trace_turns.append(new_turn)
+        for completed_turn in completed_turns:
+            self._schedule_trace_emit(completed_turn)
 
     async def attach_user_text(
         self,
@@ -450,6 +472,8 @@ class TurnTracer:
             turn.prompt_text = turn.user_transcript
             if normalized_speech_id and turn.speech_id is None:
                 turn.speech_id = normalized_speech_id
+            if normalized_speech_id:
+                self._apply_buffered_assistant_text_for_speech_id(turn)
 
             llm_attrs = _sanitize_component_attributes(metric_attributes)
             order = self._next_event_order(turn)
@@ -496,6 +520,8 @@ class TurnTracer:
 
             if normalized_speech_id and turn.speech_id is None:
                 turn.speech_id = normalized_speech_id
+            if normalized_speech_id:
+                self._apply_buffered_assistant_text_for_speech_id(turn)
 
             tts_attrs = _sanitize_component_attributes(metric_attributes)
             order = self._next_event_order(turn)
@@ -555,31 +581,47 @@ class TurnTracer:
         assistant_text: str,
         *,
         event_created_at: Optional[float] = None,
+        speech_id: Optional[str] = None,
     ) -> Optional[TraceTurn]:
         async with self._trace_lock:
-            turn = self._latest_turn_where(lambda c: bool(c.llm_calls))
-            if not turn:
-                turn = self._latest_turn_where(lambda _: True)
-            if not turn:
-                return None
             normalized_text = assistant_text.strip()
             if not normalized_text:
+                return None
+
+            normalized_speech_id = _normalize_optional_str(speech_id)
+            resolved_event_created_at = _to_optional_float(event_created_at)
+
+            if normalized_speech_id:
+                turn = self._resolve_turn_for_exact_speech_id(normalized_speech_id)
+                if turn is None:
+                    self._buffer_assistant_text(
+                        normalized_text,
+                        event_created_at=resolved_event_created_at,
+                        speech_id=normalized_speech_id,
+                    )
+                    return None
+                self._apply_assistant_text_to_turn(
+                    turn,
+                    normalized_text,
+                    event_created_at=resolved_event_created_at,
+                )
                 return turn
-            previous_assistant_text = turn.assistant_text or turn.response_text
-            order = self._next_event_order(turn)
-            turn.assistant_text = normalized_text
-            turn.response_text = normalized_text
-            assistant_event_created_at = _to_optional_float(event_created_at)
-            turn.assistant_text_updated_at = _resolved_event_timestamp(
-                assistant_event_created_at
+
+            turn = self._select_turn_for_orphan_assistant_text(
+                event_created_at=resolved_event_created_at
             )
-            turn.assistant_text_updated_order = order
-            _reconcile_assistant_text_with_tts_calls(
-                turn=turn,
-                assistant_text=normalized_text,
-                previous_assistant_text=previous_assistant_text,
+            if turn is None:
+                self._buffer_assistant_text(
+                    normalized_text,
+                    event_created_at=resolved_event_created_at,
+                )
+                return None
+
+            self._apply_assistant_text_to_turn(
+                turn,
+                normalized_text,
+                event_created_at=resolved_event_created_at,
             )
-            self._maybe_close_tool_phase(turn)
             return turn
 
     async def attach_tool_step_started(self) -> tuple[Optional[TraceTurn], bool]:
@@ -651,6 +693,55 @@ class TurnTracer:
             self._maybe_close_tool_phase(turn)
             return turn
 
+    async def mark_first_audio_started(
+        self,
+        *,
+        speech_id: str,
+        started_at: Optional[float] = None,
+    ) -> Optional[TraceTurn]:
+        async with self._trace_lock:
+            normalized_speech_id = _normalize_optional_str(speech_id)
+            if not normalized_speech_id:
+                return None
+            turn = self._resolve_turn_for_exact_speech_id(normalized_speech_id)
+            if turn is None:
+                return None
+            turn.assistant_audio_started = True
+            turn.assistant_audio_started_at = _resolved_event_timestamp(
+                _to_optional_float(started_at)
+            )
+            return turn
+
+    async def drain_pending_turns(self) -> None:
+        completed_turns: list[TraceTurn] = []
+        async with self._trace_lock:
+            for turn in list(self._pending_trace_turns):
+                self._apply_buffered_assistant_text_for_speech_id(turn)
+                self._try_attach_latest_usable_orphan_assistant_text(turn)
+                if not (turn.user_transcript and turn.llm_calls and turn.tts_calls):
+                    continue
+                requires_post_tool_response = self._requires_post_tool_follow_up(turn)
+                missing_post_tool_assistant = bool(
+                    requires_post_tool_response
+                    and not self._post_tool_assistant_observed(turn)
+                )
+                if turn.interrupted_reason is None:
+                    turn.interrupted = True
+                    turn.interrupted_reason = "shutdown_drain"
+                completed_turns.append(
+                    self._finalize_locked(
+                        turn,
+                        missing_assistant_fallback=(
+                            missing_post_tool_assistant or not bool(turn.assistant_text)
+                        ),
+                        tool_post_response_missing=requires_post_tool_response,
+                        drop_assistant_text=missing_post_tool_assistant,
+                    )
+                )
+
+        for completed_turn in completed_turns:
+            self._schedule_trace_emit(completed_turn)
+
     # ------------------------------------------------------------------
     # Finalization
     # ------------------------------------------------------------------
@@ -725,6 +816,8 @@ class TurnTracer:
             return False
         if not turn.user_transcript.strip():
             return False
+        if turn.assistant_audio_started:
+            return False
         if turn.assistant_text.strip() or turn.response_text.strip():
             return False
         if turn.tool_step_announced or turn.tool_executions or turn.last_tool_event_order is not None:
@@ -768,6 +861,13 @@ class TurnTracer:
         new_turn.coalesced_fragment_count = (
             absorbed_turn.coalesced_fragment_count + 1
         )
+        absorbed_recent_activity = self._turn_recent_activity_at(absorbed_turn)
+        existing_cutoff = new_turn.orphan_assistant_cutoff_at
+        if absorbed_recent_activity is not None:
+            new_turn.orphan_assistant_cutoff_at = max(
+                existing_cutoff or absorbed_recent_activity,
+                absorbed_recent_activity,
+            )
 
     def _next_turn_where(
         self,
@@ -813,6 +913,132 @@ class TurnTracer:
             return matched_without_id
         return self._latest_turn_where(lambda c: bool(c.llm_calls))
 
+    def _resolve_turn_for_exact_speech_id(self, speech_id: str) -> Optional[TraceTurn]:
+        matched = self._latest_turn_where(lambda c: c.speech_id == speech_id)
+        if matched is not None:
+            return matched
+        candidates = [
+            turn
+            for turn in self._pending_trace_turns
+            if turn.speech_id is None and bool(turn.llm_calls or turn.tts_calls)
+        ]
+        if len(candidates) != 1:
+            return None
+        turn = candidates[0]
+        turn.speech_id = speech_id
+        return turn
+
+    def _select_turn_for_orphan_assistant_text(
+        self,
+        *,
+        event_created_at: Optional[float],
+    ) -> Optional[TraceTurn]:
+        candidates = [
+            turn
+            for turn in self._pending_trace_turns
+            if bool(turn.llm_calls or turn.tts_calls or turn.tool_phase_open)
+        ]
+        if len(candidates) != 1:
+            return None
+        turn = candidates[0]
+        cutoff = turn.orphan_assistant_cutoff_at
+        if cutoff is not None:
+            if event_created_at is None:
+                return None
+            if event_created_at < cutoff:
+                return None
+        return turn
+
+    def _apply_assistant_text_to_turn(
+        self,
+        turn: TraceTurn,
+        assistant_text: str,
+        *,
+        event_created_at: Optional[float],
+    ) -> None:
+        previous_assistant_text = turn.assistant_text or turn.response_text
+        order = self._next_event_order(turn)
+        turn.assistant_text = assistant_text
+        turn.response_text = assistant_text
+        turn.assistant_text_updated_at = _resolved_event_timestamp(event_created_at)
+        turn.assistant_text_updated_order = order
+        _reconcile_assistant_text_with_tts_calls(
+            turn=turn,
+            assistant_text=assistant_text,
+            previous_assistant_text=previous_assistant_text,
+        )
+        self._maybe_close_tool_phase(turn)
+
+    def _buffer_assistant_text(
+        self,
+        assistant_text: str,
+        *,
+        event_created_at: Optional[float],
+        speech_id: Optional[str] = None,
+    ) -> None:
+        normalized = assistant_text.strip()
+        if not normalized:
+            return
+        record = AssistantTextRecord(
+            text=normalized,
+            event_created_at=_to_optional_float(event_created_at),
+        )
+        normalized_speech_id = _normalize_optional_str(speech_id)
+        if normalized_speech_id:
+            queue = self._pending_agent_transcripts_by_speech_id.setdefault(
+                normalized_speech_id,
+                deque(),
+            )
+            if queue and queue[-1].text == normalized:
+                return
+            queue.append(record)
+            return
+        if self._orphan_assistant_text_records and self._orphan_assistant_text_records[-1].text == normalized:
+            return
+        self._orphan_assistant_text_records.append(record)
+
+    def _apply_buffered_assistant_text_for_speech_id(self, turn: TraceTurn) -> None:
+        speech_id = _normalize_optional_str(turn.speech_id)
+        if not speech_id:
+            return
+        queue = self._pending_agent_transcripts_by_speech_id.get(speech_id)
+        if not queue:
+            return
+        while queue:
+            record = queue.popleft()
+            self._apply_assistant_text_to_turn(
+                turn,
+                record.text,
+                event_created_at=record.event_created_at,
+            )
+        if not queue:
+            self._pending_agent_transcripts_by_speech_id.pop(speech_id, None)
+
+    def _try_attach_latest_usable_orphan_assistant_text(
+        self,
+        turn: TraceTurn,
+    ) -> bool:
+        if not self._orphan_assistant_text_records:
+            return False
+        if self._select_turn_for_orphan_assistant_text(
+            event_created_at=self._orphan_assistant_text_records[-1].event_created_at
+        ) is not turn:
+            return False
+        for index in range(len(self._orphan_assistant_text_records) - 1, -1, -1):
+            record = self._orphan_assistant_text_records[index]
+            if self._select_turn_for_orphan_assistant_text(
+                event_created_at=record.event_created_at
+            ) is not turn:
+                continue
+            del self._orphan_assistant_text_records[index]
+            self._apply_assistant_text_to_turn(
+                turn,
+                record.text,
+                event_created_at=record.event_created_at,
+            )
+            return True
+        return False
+
     def _maybe_update_perceived_second_audio_latency(
         self,
         turn: TraceTurn,
@@ -985,6 +1211,7 @@ class TurnTracer:
         min_observed_order: Optional[int] = None,
         include_pending_agent_transcripts: bool = True,
     ) -> str:
+        speech_id = _normalize_optional_str(turn.speech_id)
         if turn.assistant_text.strip():
             if (
                 min_observed_order is None
@@ -1003,6 +1230,16 @@ class TurnTracer:
                 )
             ):
                 return turn.response_text.strip()
+        if speech_id:
+            buffered_exact = self._pending_agent_transcripts_by_speech_id.get(speech_id)
+            if buffered_exact:
+                while buffered_exact:
+                    record = buffered_exact.popleft()
+                    if not record.text.strip():
+                        continue
+                    if not buffered_exact:
+                        self._pending_agent_transcripts_by_speech_id.pop(speech_id, None)
+                    return record.text.strip()
         for tts_call in reversed(turn.tts_calls):
             if (
                 min_observed_order is not None
@@ -1011,10 +1248,44 @@ class TurnTracer:
                 continue
             if tts_call.assistant_text.strip():
                 return tts_call.assistant_text.strip()
+        if self._try_attach_latest_usable_orphan_assistant_text(turn):
+            if turn.assistant_text.strip():
+                return turn.assistant_text.strip()
         if include_pending_agent_transcripts and self._pending_agent_transcripts:
             return self._pending_agent_transcripts.popleft().strip()
         return ""
 
+    def _finalize_interrupted_turns_before_new_user_turn_locked(self) -> list[TraceTurn]:
+        completed_turns: list[TraceTurn] = []
+        for turn in list(self._pending_trace_turns):
+            if not (turn.user_transcript and turn.llm_calls and turn.tts_calls):
+                continue
+            if not turn.assistant_audio_started:
+                continue
+            turn.interrupted = True
+            if not turn.interrupted_reason:
+                turn.interrupted_reason = "user_barge_in_after_audio_started"
+            requires_post_tool_response = self._requires_post_tool_follow_up(turn)
+            missing_post_tool_assistant = bool(
+                requires_post_tool_response and not self._post_tool_assistant_observed(turn)
+            )
+            completed_turns.append(
+                self._finalize_locked(
+                    turn,
+                    missing_assistant_fallback=(
+                        missing_post_tool_assistant or not bool(turn.assistant_text)
+                    ),
+                    tool_post_response_missing=requires_post_tool_response,
+                    drop_assistant_text=missing_post_tool_assistant,
+                )
+            )
+        return completed_turns
+
+    def _requires_post_tool_follow_up(self, turn: TraceTurn) -> bool:
+        if turn.last_tool_event_order is None:
+            return False
+        return self._requires_post_tool_response(turn)
+
     # ------------------------------------------------------------------
     # Timeout scheduling
     # ------------------------------------------------------------------
@@ -2155,6 +2426,9 @@ def _set_root_attributes(
         "langfuse.trace.metadata.tool_phase_announced": turn.tool_step_announced,
         "langfuse.trace.metadata.tool_post_response_missing": turn.tool_post_response_missing,
         "langfuse.trace.metadata.user_turn_committed": turn.user_turn_committed,
+        "langfuse.trace.metadata.assistant_audio_started": turn.assistant_audio_started,
+        "langfuse.trace.metadata.interrupted": turn.interrupted,
+        "langfuse.trace.metadata.interrupted_reason": turn.interrupted_reason,
         "langfuse.trace.metadata.coalesced_turn_count": len(turn.coalesced_turn_ids),
         "langfuse.trace.metadata.coalesced_fragment_count": turn.coalesced_fragment_count,
         "langfuse.trace.metadata.coalesced_turn_ids": turn.coalesced_turn_ids,

src/core/settings.py

@@ -422,6 +422,15 @@ class LangfuseSettings(CoreSettings):
         le=10000.0,
         description="Best-effort tracer flush timeout in milliseconds",
     )
+    LANGFUSE_SHUTDOWN_DRAIN_TIMEOUT_MS: float = Field(
+        default=3000.0,
+        ge=0.0,
+        le=15000.0,
+        description=(
+            "Maximum time to drain pending Langfuse trace finalization/emission work "
+            "during shutdown before exiting"
+        ),
+    )
     LANGFUSE_CONTINUATION_COALESCE_WINDOW_MS: float = Field(
         default=1500.0,
         ge=0.0,

tests/test_langfuse_turn_tracing.py

@@ -1758,6 +1758,7 @@ def test_immediate_continuation_coalesces_aborted_prior_turn(
     )
 
     async def _run() -> None:
+        base_time = time.time()
         await collector.on_session_metadata(
             session_id="session-coalesce",
             participant_id="web-123",
@@ -1779,9 +1780,16 @@ def test_immediate_continuation_coalesces_aborted_prior_turn(
         await collector.on_conversation_item_added(
             role="user",
             content="What the difference between speech to text and speech recognition?",
+            event_created_at=base_time + 0.2,
+            item_created_at=base_time + 0.2,
         )
         await collector.on_metrics_collected(_make_llm_metrics("speech-b"))
-        await collector.on_conversation_item_added(role="assistant", content="Speech to text writes words down.")
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="Speech to text writes words down.",
+            event_created_at=base_time + 0.4,
+            item_created_at=base_time + 0.4,
+        )
         await collector.on_metrics_collected(_make_tts_metrics("speech-b"))
         await collector.wait_for_pending_trace_tasks()
 
@@ -2110,7 +2118,10 @@ def test_speech_created_done_callback_backfills_assistant_text(
         await collector.wait_for_pending_trace_tasks()
         assert not fake_tracer.spans
 
-        handle = _FakeSpeechHandle(chat_items=[_FakeChatItem(role="assistant", content=["fallback reply"])])
+        handle = _FakeSpeechHandle(
+            chat_items=[_FakeChatItem(role="assistant", content=["fallback reply"])],
+            speech_id="speech-speech-created",
+        )
         await collector.on_speech_created(handle)
         handle.trigger_done()
         await asyncio.sleep(0)
@@ -2159,7 +2170,8 @@ def test_speech_created_immediate_capture_backfills_assistant_text(
                     role="assistant",
                     content=[_FakeTextMethodPart("immediate fallback reply")],
                 )
-            ]
+            ],
+            speech_id="speech-speech-created-immediate",
         )
         await collector.on_speech_created(handle)
         await collector.wait_for_pending_trace_tasks()
@@ -2480,3 +2492,239 @@ def test_real_participant_metadata_overrides_fallback_for_pending_turns(
     assert len(turn_spans) == 1
     assert turn_spans[0].attributes["session_id"] == "session-real-participant"
     assert turn_spans[0].attributes["participant_id"] == "web-real-participant"
+
+
+def test_stale_orphan_assistant_text_from_absorbed_turn_is_not_attached_to_continuation(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.metrics_collector as metrics_collector_module
+
+    fake_tracer = _FakeTracer()
+    monkeypatch.setattr(metrics_collector_module, "tracer", fake_tracer)
+
+    room = _FakeRoom()
+    collector = MetricsCollector(
+        room=room,  # type: ignore[arg-type]
+        model_name="moonshine",
+        room_name=room.name,
+        room_id="RM123",
+        participant_id="web-123",
+        langfuse_enabled=True,
+    )
+
+    base_time = time.time()
+    final_input = "What is the difference between speech to text and speech recognition?"
+    final_output = (
+        "Speech recognition detects spoken words. Speech to text writes them down."
+    )
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-stale-orphan-after-coalesce",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("What", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-a", delay=0.7, transcription_delay=0.2)
+        )
+        await collector.on_metrics_collected(_make_llm_metrics("speech-a"))
+        await collector.on_metrics_collected(_make_tts_metrics("speech-a"))
+
+        await collector.on_user_input_transcribed(
+            "is the difference between speech to text and speech recognition?",
+            is_final=True,
+        )
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-b", delay=0.7, transcription_delay=0.2)
+        )
+        await collector.on_conversation_item_added(
+            role="user",
+            content=final_input,
+            event_created_at=base_time + 0.25,
+            item_created_at=base_time + 0.25,
+        )
+        await collector.on_metrics_collected(_make_llm_metrics("speech-b"))
+
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="stale reply from the absorbed turn",
+            event_created_at=base_time + 1.0,
+            item_created_at=base_time + 0.1,
+        )
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content=final_output,
+            event_created_at=base_time + 1.2,
+            item_created_at=base_time + 1.2,
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-b"))
+        await collector.wait_for_pending_trace_tasks()
+
+    asyncio.run(_run())
+
+    turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+    assert len(turn_spans) == 1
+    root = turn_spans[0]
+    assert root.attributes["langfuse.trace.input"] == final_input
+    assert root.attributes["langfuse.trace.output"] == final_output
+    assert "stale reply from the absorbed turn" not in root.attributes["langfuse.trace.output"]
+
+
+def test_audio_started_turn_is_finalized_separately_when_new_user_turn_arrives(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.metrics_collector as metrics_collector_module
+
+    fake_tracer = _FakeTracer()
+    monkeypatch.setattr(metrics_collector_module, "tracer", fake_tracer)
+
+    room = _FakeRoom()
+    collector = MetricsCollector(
+        room=room,  # type: ignore[arg-type]
+        model_name="moonshine",
+        room_name=room.name,
+        room_id="RM123",
+        participant_id="web-123",
+        langfuse_enabled=True,
+    )
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-audio-started-no-coalesce",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("first prompt", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-audio-started", delay=0.4, transcription_delay=0.1)
+        )
+        await collector.on_metrics_collected(_make_llm_metrics("speech-audio-started"))
+        await collector.on_speech_created(
+            _FakeSpeechHandle(chat_items=[], speech_id="speech-audio-started")
+        )
+        await collector.on_agent_state_changed(
+            old_state="thinking",
+            new_state="speaking",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-audio-started"))
+
+        await collector.on_user_input_transcribed("second prompt", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-separate-b", delay=0.5, transcription_delay=0.1)
+        )
+        await collector.on_metrics_collected(_make_llm_metrics("speech-separate-b"))
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="second reply",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-separate-b"))
+        await collector.wait_for_pending_trace_tasks()
+
+    asyncio.run(_run())
+
+    turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+    assert len(turn_spans) == 2
+    first, second = turn_spans
+    assert first.attributes["langfuse.trace.input"] == "first prompt"
+    assert first.attributes["langfuse.trace.metadata.interrupted"] is True
+    assert first.attributes["langfuse.trace.output"] == "[assistant text unavailable]"
+    assert second.attributes["langfuse.trace.input"] == "second prompt"
+    assert second.attributes["langfuse.trace.output"] == "second reply"
+
+
+def test_interrupted_pretool_leadin_keeps_own_output_and_next_turn_stays_separate(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.metrics_collector as metrics_collector_module
+
+    fake_tracer = _FakeTracer()
+    monkeypatch.setattr(metrics_collector_module, "tracer", fake_tracer)
+
+    room = _FakeRoom()
+    collector = MetricsCollector(
+        room=room,  # type: ignore[arg-type]
+        model_name="moonshine",
+        room_name=room.name,
+        room_id="RM123",
+        participant_id="web-123",
+        langfuse_enabled=True,
+    )
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-interrupted-pretool-leadin",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("search papers", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-pretool-interrupted"))
+        await collector.on_tool_step_started()
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="Let me check that.",
+        )
+        await collector.on_speech_created(
+            _FakeSpeechHandle(chat_items=[], speech_id="speech-pretool-interrupted")
+        )
+        await collector.on_agent_state_changed(
+            old_state="thinking",
+            new_state="speaking",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-pretool-interrupted"))
+
+        await collector.on_user_input_transcribed("never mind", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-pretool-next"))
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="Okay, stopping here.",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-pretool-next"))
+        await collector.wait_for_pending_trace_tasks()
+
+    asyncio.run(_run())
+
+    turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+    assert len(turn_spans) == 2
+    first, second = turn_spans
+    assert first.attributes["langfuse.trace.input"] == "search papers"
+    assert first.attributes["langfuse.trace.metadata.interrupted"] is True
+    assert first.attributes["langfuse.trace.output"] == "Let me check that."
+    assert second.attributes["langfuse.trace.input"] == "never mind"
+    assert second.attributes["langfuse.trace.output"] == "Okay, stopping here."
+
+
+def test_drain_pending_traces_finalizes_without_manual_sleep(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import src.agent.traces.metrics_collector as metrics_collector_module
+
+    fake_tracer = _FakeTracer()
+    monkeypatch.setattr(metrics_collector_module, "tracer", fake_tracer)
+
+    room = _FakeRoom()
+    collector = MetricsCollector(
+        room=room,  # type: ignore[arg-type]
+        model_name="moonshine",
+        room_name=room.name,
+        room_id="RM123",
+        participant_id="web-123",
+        langfuse_enabled=True,
+    )
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-drain-pending-traces",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("drain pending", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-drain-pending"))
+        await collector.on_metrics_collected(_make_tts_metrics("speech-drain-pending"))
+        await collector.drain_pending_traces()
+
+    asyncio.run(_run())
+
+    turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+    assert len(turn_spans) == 1
+    root = turn_spans[0]
+    assert root.attributes["langfuse.trace.input"] == "drain pending"
+    assert root.attributes["langfuse.trace.output"] == "[assistant text unavailable]"
+    assert root.attributes["langfuse.trace.metadata.interrupted"] is True
+    assert root.attributes["langfuse.trace.metadata.interrupted_reason"] == "shutdown_drain"

tests/test_runtime_settings.py

@@ -56,6 +56,7 @@ def test_langfuse_runtime_tuning_defaults_are_declared() -> None:
 
     assert fields["LANGFUSE_TRACE_FINALIZE_TIMEOUT_MS"].default == 8000.0
     assert fields["LANGFUSE_POST_TOOL_RESPONSE_TIMEOUT_MS"].default == 30000.0
+    assert fields["LANGFUSE_SHUTDOWN_DRAIN_TIMEOUT_MS"].default == 3000.0
     assert fields["LANGFUSE_CONTINUATION_COALESCE_WINDOW_MS"].default == 1500.0