feat: Enhance turn tracing and metrics collection

- Introduced `LANGFUSE_ASSISTANT_TEXT_GRACE_TIMEOUT_MS` in `.env.example` and `settings.py` to manage the grace period for waiting on assistant text before finalizing traces.
- Refactored `MetricsCollector` and `TurnTracer` to support new metrics and improve handling of user utterances and assistant text sources.
- Updated `PocketTTS` initialization to streamline parameter handling.
- Enhanced tests to validate the new features and ensure proper functionality of the updated tracing and metrics collection mechanisms.

.env.example

@@ -42,7 +42,8 @@ LANGFUSE_PROJECT_ID=  # Required for frontend deep links: project/<project_id>/.
 LANGFUSE_PUBLIC_KEY=
 LANGFUSE_SECRET_KEY=
 LANGFUSE_PUBLIC_TRACES=false  # Mark traces public so non-members can open shared links
-LANGFUSE_TRACE_FINALIZE_TIMEOUT_MS=8000
+LANGFUSE_ASSISTANT_TEXT_GRACE_TIMEOUT_MS=500  # Short wait for assistant text on normal turns
+LANGFUSE_TRACE_FINALIZE_TIMEOUT_MS=8000  # Legacy fallback retained for compatibility
 LANGFUSE_POST_TOOL_RESPONSE_TIMEOUT_MS=30000
 LANGFUSE_MAX_PENDING_TRACE_TASKS=200
 LANGFUSE_TRACE_FLUSH_TIMEOUT_MS=1000

src/agent/models/tts_factory.py

@@ -127,11 +127,12 @@ def create_tts() -> Any:
             settings.voice.POCKET_TTS_TEMPERATURE,
             settings.voice.POCKET_TTS_LSD_DECODE_STEPS,
         )
-        return PocketTTS(
+        pocket_kwargs: dict[str, Any] = dict(
             voice=settings.voice.POCKET_TTS_VOICE,
             temperature=settings.voice.POCKET_TTS_TEMPERATURE,
             lsd_decode_steps=settings.voice.POCKET_TTS_LSD_DECODE_STEPS,
         )
+        return PocketTTS(**pocket_kwargs)
 
     if provider == "deepgram":
         logger.info("Initializing Deepgram TTS with plugin defaults")

src/agent/traces/metrics_collector.py

@@ -7,6 +7,7 @@ real-time monitoring. Also creates one Langfuse trace per finalized user turn.
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import uuid
 from collections import deque
@@ -211,6 +212,7 @@ class PendingUserUtterance:
     stt_observed: bool = False
     llm_started: bool = False
     watchdog_id: Optional[str] = None
+    speech_id: Optional[str] = None
 
 
 @dataclass
@@ -271,6 +273,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._speech_item_callback_registered_logged = False
+        self._speech_item_callback_unavailable_logged = False
+        self._speech_item_callback_failed_logged = False
         self._event_queue: deque[QueuedCollectorEvent] = deque()
         self._event_worker_task: asyncio.Task[None] | None = None
         self._event_loop: asyncio.AbstractEventLoop | None = None
@@ -492,7 +497,7 @@ class MetricsCollector:
         if not normalized:
             return
         if role == "user":
-            utterance = self._latest_user_utterance()
+            utterance = self._user_utterance_accepting_manual_update()
             if utterance is None:
                 utterance = PendingUserUtterance(
                     transcript=normalized,
@@ -500,7 +505,10 @@ class MetricsCollector:
                 )
                 self._pending_user_utterances.append(utterance)
             else:
-                utterance.transcript = normalized
+                utterance.transcript = _merge_user_transcripts(
+                    utterance.transcript,
+                    normalized,
+                )
                 utterance.committed = True
                 if utterance.watchdog_id is not None:
                     self._update_llm_stall_watchdog(
@@ -511,8 +519,9 @@ class MetricsCollector:
                 item_created_at if item_created_at is not None else event_created_at
             )
             await self._tracer.attach_user_text(
-                normalized,
+                utterance.transcript,
                 event_created_at=user_event_created_at,
+                speech_id=utterance.speech_id,
             )
             return
         assistant_event_created_at = (
@@ -521,6 +530,7 @@ class MetricsCollector:
         await self._on_assistant_text(
             normalized,
             event_created_at=assistant_event_created_at,
+            source="conversation_item",
         )
 
     async def on_function_tools_executed(
@@ -569,6 +579,11 @@ class MetricsCollector:
         if speech_id:
             self._pending_speech_ids_for_first_audio.append(speech_id)
 
+        on_item_added = self._register_speech_item_added_callback(
+            speech_handle=speech_handle,
+            speech_id=speech_id,
+        )
+
         assistant_text, assistant_created_at = _extract_latest_assistant_chat_item(
             getattr(speech_handle, "chat_items", [])
         )
@@ -577,6 +592,7 @@ class MetricsCollector:
                 assistant_text,
                 event_created_at=assistant_created_at,
                 speech_id=speech_id,
+                source="speech_created",
             )
 
         add_done_callback = getattr(speech_handle, "add_done_callback", None)
@@ -584,25 +600,45 @@ class MetricsCollector:
             return
 
         def _on_done(handle: Any) -> None:
-            try:
-                done_speech_id = _normalize(getattr(handle, "id", None))
-                text, created_at = _extract_latest_assistant_chat_item(
-                    getattr(handle, "chat_items", [])
-                )
-                self._submit_serialized(
-                    self._handle_speech_done,
-                    done_speech_id,
-                    text,
-                    created_at,
-                )
-            except Exception:
-                return
+            remove_item_added_callback = getattr(
+                handle,
+                "_remove_item_added_callback",
+                None,
+            )
+            if callable(remove_item_added_callback) and on_item_added is not None:
+                with contextlib.suppress(Exception):
+                    remove_item_added_callback(on_item_added)
+            done_speech_id = _normalize(getattr(handle, "id", None))
+            text, created_at = _extract_latest_assistant_chat_item(
+                getattr(handle, "chat_items", [])
+            )
+            self._submit_serialized_callback(
+                self._handle_speech_done,
+                done_speech_id,
+                text,
+                created_at,
+            )
 
         try:
             add_done_callback(_on_done)
         except Exception:
             return
 
+    async def _handle_speech_item_added(
+        self,
+        speech_id: Optional[str],
+        assistant_text: str,
+        event_created_at: Optional[float],
+    ) -> None:
+        if not assistant_text:
+            return
+        await self._on_assistant_text(
+            assistant_text,
+            event_created_at=event_created_at,
+            speech_id=speech_id,
+            source="speech_item_added",
+        )
+
     async def _handle_speech_done(
         self,
         speech_id: Optional[str],
@@ -616,6 +652,7 @@ class MetricsCollector:
                 assistant_text,
                 event_created_at=event_created_at,
                 speech_id=speech_id,
+                source="speech_done",
             )
 
     async def on_agent_state_changed(
@@ -669,70 +706,6 @@ class MetricsCollector:
                 started_at=time(),
             )
 
-    async def on_tts_synthesized(
-        self,
-        *,
-        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
-        speech_id = self._latest_agent_speech_id or f"tts-{uuid.uuid4()}"
-        turn_metrics = self._get_or_create_turn(speech_id, role="agent")
-        turn_metrics.tts = TTSMetrics(
-            type="tts_metrics",
-            label="tts_fallback",
-            request_id=f"fallback-{speech_id}",
-            timestamp=time(),
-            duration=duration,
-            ttfb=ttfb,
-            audio_duration=audio_duration,
-            cancelled=False,
-            characters_count=0,
-            streamed=True,
-            speech_id=speech_id,
-        )
-        await self._publish_live_update(speech_id=speech_id, stage="tts", turn_metrics=turn_metrics)
-        logger.debug("TTS fallback metrics collected: speech_id=%s, ttfb=%.3fs", speech_id, ttfb)
-        await self._maybe_publish_turn(speech_id, turn_metrics)
-
-        trace_turn = await self._tracer.attach_tts(
-            duration=duration,
-            fallback_duration=audio_duration,
-            ttfb=ttfb,
-            speech_id=speech_id,
-            observed_total_latency=self._observed_total_latency(speech_id),
-            metric_attributes={
-                "type": "tts_metrics",
-                "label": "tts_fallback",
-                "request_id": f"fallback-{speech_id}",
-                "timestamp": time(),
-                "duration": duration,
-                "ttfb": ttfb,
-                "audio_duration": audio_duration,
-                "cancelled": False,
-                "characters_count": 0,
-                "streamed": True,
-                "speech_id": speech_id,
-            },
-        )
-        await self._tracer.maybe_finalize(trace_turn)
-
     async def on_metrics_collected(
         self,
         collected_metrics: Union[
@@ -790,10 +763,12 @@ class MetricsCollector:
             )
 
         elif isinstance(collected_metrics, metrics.LLMMetrics):
-            self._mark_llm_stage_reached()
             speech_id = collected_metrics.speech_id or collected_metrics.request_id
+            linked_utterance = self._mark_llm_stage_reached(speech_id)
             turn_metrics = self._get_or_create_turn(speech_id, role="agent")
             self._latest_agent_speech_id = speech_id
+            if linked_utterance is not None:
+                turn_metrics.transcript = linked_utterance.transcript
             turn_metrics.llm = LLMMetrics(
                 type=collected_metrics.type,
                 label=collected_metrics.label,
@@ -822,6 +797,7 @@ class MetricsCollector:
         elif isinstance(collected_metrics, metrics.TTSMetrics):
             speech_id = collected_metrics.speech_id or collected_metrics.request_id
             turn_metrics = self._get_or_create_turn(speech_id, role="agent")
+            tts_metric_metadata = _metric_metadata_to_dict(collected_metrics.metadata)
             turn_metrics.tts = TTSMetrics(
                 type=collected_metrics.type,
                 label=collected_metrics.label,
@@ -835,7 +811,7 @@ class MetricsCollector:
                 streamed=collected_metrics.streamed,
                 segment_id=collected_metrics.segment_id,
                 speech_id=collected_metrics.speech_id,
-                metadata=_metric_metadata_to_dict(collected_metrics.metadata),
+                metadata=tts_metric_metadata,
             )
             await self._publish_live_update(speech_id=speech_id, stage="tts", turn_metrics=turn_metrics)
             logger.debug("TTS metrics collected: speech_id=%s, ttfb=%.3fs", speech_id, collected_metrics.ttfb)
@@ -847,11 +823,21 @@ class MetricsCollector:
                 observed_total_latency=self._observed_total_latency(speech_id),
                 metric_attributes=_tts_metric_attributes(collected_metrics),
             )
+            metric_assistant_text = _assistant_text_from_metadata(tts_metric_metadata)
+            if metric_assistant_text:
+                await self._on_assistant_text(
+                    metric_assistant_text,
+                    event_created_at=collected_metrics.timestamp,
+                    speech_id=speech_id,
+                    source="tts_metrics",
+                )
 
         elif isinstance(collected_metrics, metrics.EOUMetrics):
             speech_id = collected_metrics.speech_id
             if speech_id:
-                self._mark_oldest_open_user_utterance_committed()
+                linked_utterance = self._mark_oldest_open_user_utterance_committed(
+                    speech_id
+                )
                 state = self._get_or_create_state(speech_id)
                 if state.speech_end_monotonic is None:
                     state.speech_end_monotonic = monotonic()
@@ -867,6 +853,8 @@ class MetricsCollector:
                     metadata=_metric_metadata_to_dict(collected_metrics.metadata),
                 )
                 turn_metrics = state.metrics
+                if turn_metrics and linked_utterance is not None:
+                    turn_metrics.transcript = linked_utterance.transcript
                 if turn_metrics:
                     turn_metrics.eou = state.eou_metrics
                     if self._latest_vad_metrics and turn_metrics.vad is None:
@@ -1037,6 +1025,7 @@ class MetricsCollector:
         *,
         event_created_at: Optional[float] = None,
         speech_id: Optional[str] = None,
+        source: str = "unknown",
     ) -> None:
         normalized = assistant_text.strip()
         if not normalized:
@@ -1045,9 +1034,86 @@ class MetricsCollector:
             normalized,
             event_created_at=event_created_at,
             speech_id=speech_id,
+            source=source,
         )
         await self._tracer.maybe_finalize(trace_turn)
 
+    def _register_speech_item_added_callback(
+        self,
+        *,
+        speech_handle: Any,
+        speech_id: Optional[str],
+    ) -> Callable[[Any], None] | None:
+        add_item_added_callback = getattr(
+            speech_handle,
+            "_add_item_added_callback",
+            None,
+        )
+        if not callable(add_item_added_callback):
+            if not self._speech_item_callback_unavailable_logged:
+                self._speech_item_callback_unavailable_logged = True
+                logger.warning(
+                    "SpeechHandle item-added callback unavailable; Langfuse tracing will rely on fallback sources"
+                )
+            return None
+
+        def _on_item_added(item: Any) -> None:
+            try:
+                assistant_text, created_at = _extract_assistant_chat_item(item)
+                if not assistant_text:
+                    return
+                self._submit_serialized_callback(
+                    self._handle_speech_item_added,
+                    speech_id,
+                    assistant_text,
+                    created_at,
+                )
+            except Exception:
+                return
+
+        try:
+            add_item_added_callback(_on_item_added)
+        except Exception as exc:
+            if not self._speech_item_callback_failed_logged:
+                self._speech_item_callback_failed_logged = True
+                logger.warning(
+                    "Failed to register SpeechHandle item-added callback; Langfuse tracing will rely on fallback sources: %s",
+                    exc,
+                )
+            return None
+
+        if not self._speech_item_callback_registered_logged:
+            self._speech_item_callback_registered_logged = True
+            logger.debug(
+                "SpeechHandle item-added callback registered for provider-agnostic assistant text capture"
+            )
+        return _on_item_added
+
+    def _submit_serialized_callback(
+        self,
+        handler: Callable[..., Awaitable[Any]],
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        loop = self._event_loop
+        if loop is None or loop.is_closed():
+            return
+
+        def _enqueue() -> None:
+            self._enqueue_serialized(handler, args=args, kwargs=kwargs, waiter=None)
+
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop.call_soon_threadsafe(_enqueue)
+            return
+
+        if running_loop is loop:
+            _enqueue()
+            return
+
+        loop.call_soon_threadsafe(_enqueue)
+
     async def _publish_live_update(
         self,
         *,
@@ -1124,7 +1190,17 @@ class MetricsCollector:
 
     def _current_open_user_utterance(self) -> Optional[PendingUserUtterance]:
         utterance = self._latest_user_utterance()
-        if utterance is None or utterance.committed or utterance.llm_started:
+        if utterance is None or utterance.llm_started:
+            return None
+        if utterance.committed and utterance.speech_id is None:
+            return None
+        return utterance
+
+    def _user_utterance_accepting_manual_update(self) -> Optional[PendingUserUtterance]:
+        utterance = self._latest_user_utterance()
+        if utterance is None or utterance.llm_started:
+            return None
+        if utterance.committed and utterance.speech_id is None:
             return None
         return utterance
 
@@ -1140,13 +1216,40 @@ class MetricsCollector:
             return utterance
         return None
 
-    def _mark_oldest_open_user_utterance_committed(self) -> None:
+    def _find_user_utterance_by_speech_id(
+        self,
+        speech_id: str,
+        *,
+        include_llm_started: bool = False,
+    ) -> Optional[PendingUserUtterance]:
+        for utterance in reversed(self._pending_user_utterances):
+            if utterance.speech_id != speech_id:
+                continue
+            if utterance.llm_started and not include_llm_started:
+                continue
+            return utterance
+        return None
+
+    def _mark_oldest_open_user_utterance_committed(
+        self,
+        speech_id: str,
+    ) -> Optional[PendingUserUtterance]:
+        linked = self._find_user_utterance_by_speech_id(speech_id)
+        if linked is not None:
+            linked.committed = True
+            self._prune_resolved_user_utterances()
+            return linked
+
         for utterance in self._pending_user_utterances:
-            if utterance.committed:
+            if utterance.llm_started:
+                continue
+            if utterance.speech_id is not None:
                 continue
             utterance.committed = True
+            utterance.speech_id = speech_id
             self._prune_resolved_user_utterances()
-            return
+            return utterance
+        return None
 
     def _prune_resolved_user_utterances(self) -> None:
         while self._pending_user_utterances:
@@ -1181,19 +1284,36 @@ class MetricsCollector:
             return
         utterance.transcript = transcript
 
-    def _mark_llm_stage_reached(self) -> None:
-        for utterance in self._pending_user_utterances:
-            if utterance.llm_started:
-                continue
-            utterance.llm_started = True
-            watchdog_id = utterance.watchdog_id
-            utterance.watchdog_id = None
-            if watchdog_id is not None:
-                task = self._llm_stall_tasks.pop(watchdog_id, None)
-                if task:
-                    task.cancel()
-            self._prune_resolved_user_utterances()
-            return
+    def _mark_llm_stage_reached(
+        self,
+        speech_id: Optional[str],
+    ) -> Optional[PendingUserUtterance]:
+        normalized_speech_id = _normalize(speech_id)
+        utterance: Optional[PendingUserUtterance] = None
+        if normalized_speech_id is not None:
+            utterance = self._find_user_utterance_by_speech_id(normalized_speech_id)
+
+        if utterance is None:
+            for candidate in self._pending_user_utterances:
+                if candidate.llm_started:
+                    continue
+                utterance = candidate
+                break
+
+        if utterance is None:
+            return None
+
+        if normalized_speech_id is not None and utterance.speech_id is None:
+            utterance.speech_id = normalized_speech_id
+        utterance.llm_started = True
+        watchdog_id = utterance.watchdog_id
+        utterance.watchdog_id = None
+        if watchdog_id is not None:
+            task = self._llm_stall_tasks.pop(watchdog_id, None)
+            if task:
+                task.cancel()
+        self._prune_resolved_user_utterances()
+        return utterance
 
     def _find_user_utterance_by_watchdog(
         self,
@@ -1278,6 +1398,10 @@ def _merge_user_transcripts(existing: str, incoming: str) -> str:
         return left
     if left == right:
         return left
+    if left.casefold() in right.casefold():
+        return right
+    if right.casefold() in left.casefold():
+        return left
     if right.startswith(left):
         return right
     if left.startswith(right):
@@ -1367,6 +1491,16 @@ def _extract_latest_assistant_chat_item(chat_items: Any) -> tuple[str, Optional[
     return latest_text, latest_created_at
 
 
+def _extract_assistant_chat_item(item: Any) -> tuple[str, Optional[float]]:
+    role = getattr(item, "role", None)
+    if isinstance(role, str) and role != "assistant":
+        return "", None
+    normalized = _extract_content_text(getattr(item, "content", None)).strip()
+    if not normalized:
+        return "", None
+    return normalized, _to_optional_float(getattr(item, "created_at", None))
+
+
 def _to_optional_float(value: Any) -> Optional[float]:
     if isinstance(value, bool):
         return None
@@ -1393,6 +1527,16 @@ def _metric_metadata_to_dict(metadata: Any) -> Optional[dict[str, Any]]:
     return {"value": str(metadata)}
 
 
+def _assistant_text_from_metadata(metadata: Optional[dict[str, Any]]) -> str:
+    if not metadata:
+        return ""
+    for key in ("assistant_text", "spoken_text"):
+        value = metadata.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return ""
+
+
 def _metadata_attributes(metadata: Any) -> dict[str, Any]:
     data = _metric_metadata_to_dict(metadata)
     if not data:

src/agent/traces/turn_tracer.py

@@ -80,6 +80,9 @@ class TraceTurn:
     assistant_audio_started_at: Optional[float] = None
     interrupted: bool = False
     interrupted_reason: Optional[str] = None
+    finalization_reason: Optional[str] = None
+    assistant_text_source: Optional[str] = None
+    emit_ready_at: Optional[float] = 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)
@@ -141,6 +144,7 @@ class AssistantTextRecord:
 
     text: str
     event_created_at: Optional[float] = None
+    source: Optional[str] = None
 
 
 @dataclass
@@ -170,6 +174,7 @@ class ToolExecutionBlock:
 
 
 _DEFAULT_TRACE_FINALIZE_TIMEOUT_MS = 8000.0
+_DEFAULT_ASSISTANT_TEXT_GRACE_TIMEOUT_MS = 500.0
 _DEFAULT_POST_TOOL_RESPONSE_TIMEOUT_MS = 30000.0
 _DEFAULT_MAX_PENDING_TRACE_TASKS = 200
 _DEFAULT_TRACE_FLUSH_TIMEOUT_SEC = 1.0
@@ -215,7 +220,21 @@ class TurnTracer:
         self._trace_finalize_tasks: dict[str, asyncio.Task[None]] = {}
         self._trace_finalize_task_versions: dict[str, int] = {}
 
+        assistant_text_grace_timeout_ms = float(
+            getattr(
+                settings.langfuse,
+                "LANGFUSE_ASSISTANT_TEXT_GRACE_TIMEOUT_MS",
+                _DEFAULT_ASSISTANT_TEXT_GRACE_TIMEOUT_MS,
+            )
+        )
         self._trace_finalize_timeout_sec = (
+            max(
+                assistant_text_grace_timeout_ms,
+                0.0,
+            )
+            / 1000.0
+        )
+        self._trace_legacy_finalize_timeout_sec = (
             max(
                 getattr(
                     settings.langfuse,
@@ -323,14 +342,20 @@ class TurnTracer:
 
     async def create_turn(self, *, user_transcript: str, room_id: str) -> None:
         completed_turns: list[TraceTurn] = []
+        timeout_schedules: list[tuple[str, float]] = []
         async with self._trace_lock:
             normalized = user_transcript.strip()
             if not normalized:
                 return
 
-            completed_turns = self._finalize_interrupted_turns_before_new_user_turn_locked()
+            (
+                completed_turns,
+                timeout_schedules,
+            ) = self._finalize_interrupted_turns_before_new_user_turn_locked()
 
-            current_turn = self._latest_turn_where(lambda c: not c.user_turn_committed)
+            current_turn = self._latest_turn_where(
+                self._turn_accepting_additional_user_input
+            )
             if current_turn is not None:
                 self._update_user_turn_text(current_turn, normalized)
             else:
@@ -354,15 +379,38 @@ class TurnTracer:
 
         for completed_turn in completed_turns:
             self._schedule_trace_emit(completed_turn)
+        for turn_id, timeout_sec in timeout_schedules:
+            self._schedule_finalize_timeout(turn_id, timeout_sec)
 
     async def attach_user_text(
         self,
         user_transcript: 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: not c.assistant_text.strip())
+            normalized_speech_id = _normalize_optional_str(speech_id)
+            turn: Optional[TraceTurn] = None
+            if normalized_speech_id:
+                turn = self._latest_turn_where(
+                    lambda c: c.speech_id == normalized_speech_id
+                )
+                if turn is None:
+                    turn = self._latest_turn_where(
+                        lambda c: c.speech_id is None and not c.llm_calls
+                    )
+                    if turn is not None:
+                        turn.speech_id = normalized_speech_id
+                if turn is not None:
+                    self._absorb_pending_pre_llm_turns(turn)
+
+            if turn is None:
+                turn = self._latest_turn_where(
+                    self._turn_accepting_additional_user_input
+                )
+            if turn is None:
+                turn = self._latest_turn_where(lambda c: not c.assistant_text.strip())
             if turn is None:
                 turn = self._latest_turn_where(lambda _: True)
             if turn is None:
@@ -377,6 +425,7 @@ class TurnTracer:
                 normalized,
                 event_created_at=event_created_at,
             )
+            self._maybe_mark_emit_ready(turn)
             turn.user_turn_committed = True
             turn.user_turn_committed_at = _resolved_event_timestamp(
                 _to_optional_float(event_created_at)
@@ -499,6 +548,7 @@ class TurnTracer:
             if turn.llm_ttft_ms is None:
                 turn.llm_ttft_ms = llm_call.ttft_ms
             turn.llm_total_latency_ms = _sum_llm_duration_ms(turn.llm_calls)
+            self._maybe_mark_emit_ready(turn)
             _recompute_perceived_first_audio_latency(turn)
             return turn
 
@@ -524,6 +574,17 @@ class TurnTracer:
                 self._apply_buffered_assistant_text_for_speech_id(turn)
 
             tts_attrs = _sanitize_component_attributes(metric_attributes)
+            tts_metric_assistant_text = _assistant_text_from_component_attributes(tts_attrs)
+            if tts_metric_assistant_text and (
+                not turn.assistant_text.strip()
+                or turn.assistant_text_source == "tts_metrics"
+            ):
+                self._apply_assistant_text_to_turn(
+                    turn,
+                    tts_metric_assistant_text,
+                    event_created_at=_to_optional_float(tts_attrs.get("timestamp")),
+                    source="tts_metrics",
+                )
             order = self._next_event_order(turn)
             tts_call = TTSCallTrace(
                 duration_ms=_duration_to_ms(duration, fallback_duration),
@@ -557,6 +618,7 @@ class TurnTracer:
             turn.tts_updated_at = _resolved_event_timestamp(tts_event_created_at)
             turn.tts_updated_order = order
 
+            self._maybe_mark_emit_ready(turn)
             _recompute_perceived_first_audio_latency(turn)
             if observed_total_latency is not None and len(turn.tts_calls) == 1:
                 observed_ms = observed_total_latency * 1000.0
@@ -582,6 +644,7 @@ class TurnTracer:
         *,
         event_created_at: Optional[float] = None,
         speech_id: Optional[str] = None,
+        source: Optional[str] = None,
     ) -> Optional[TraceTurn]:
         async with self._trace_lock:
             normalized_text = assistant_text.strip()
@@ -598,12 +661,26 @@ class TurnTracer:
                         normalized_text,
                         event_created_at=resolved_event_created_at,
                         speech_id=normalized_speech_id,
+                        source=source,
                     )
                     return None
+                if (
+                    source == "tts_metrics"
+                    and turn.assistant_text.strip()
+                    and turn.assistant_text_source not in {None, "tts_metrics", "unavailable"}
+                ):
+                    return turn
+                if (
+                    source == "speech_done"
+                    and turn.assistant_text.strip()
+                    and turn.assistant_text_source == "speech_item_added"
+                ):
+                    return turn
                 self._apply_assistant_text_to_turn(
                     turn,
                     normalized_text,
                     event_created_at=resolved_event_created_at,
+                    source=source,
                 )
                 return turn
 
@@ -614,13 +691,21 @@ class TurnTracer:
                 self._buffer_assistant_text(
                     normalized_text,
                     event_created_at=resolved_event_created_at,
+                    source=source,
                 )
                 return None
 
+            if (
+                source == "tts_metrics"
+                and turn.assistant_text.strip()
+                and turn.assistant_text_source not in {None, "tts_metrics", "unavailable"}
+            ):
+                return turn
             self._apply_assistant_text_to_turn(
                 turn,
                 normalized_text,
                 event_created_at=resolved_event_created_at,
+                source=source,
             )
             return turn
 
@@ -891,15 +976,88 @@ class TurnTracer:
         turn.event_counter += 1
         return turn.event_counter
 
+    def _is_emit_ready(self, turn: TraceTurn) -> bool:
+        return bool(turn.user_transcript and turn.llm_calls and turn.tts_calls)
+
+    def _maybe_mark_emit_ready(self, turn: TraceTurn) -> None:
+        if turn.emit_ready_at is not None:
+            return
+        if not self._is_emit_ready(turn):
+            return
+        turn.emit_ready_at = time()
+
+    def _turn_accepting_additional_user_input(self, turn: TraceTurn) -> bool:
+        if turn.llm_calls or turn.tts_calls:
+            return False
+        if turn.assistant_text.strip() or turn.response_text.strip():
+            return False
+        if turn.interrupted:
+            return False
+        if not turn.user_turn_committed:
+            return True
+        return turn.speech_id is not None
+
+    def _absorb_pending_pre_llm_turns(self, anchor_turn: TraceTurn) -> None:
+        if anchor_turn not in self._pending_trace_turns:
+            return
+        try:
+            anchor_index = self._pending_trace_turns.index(anchor_turn)
+        except ValueError:
+            return
+
+        anchor_speech_id = _normalize_optional_str(anchor_turn.speech_id)
+        absorbed_turns: list[TraceTurn] = []
+        for candidate in list(self._pending_trace_turns)[anchor_index + 1 :]:
+            candidate_speech_id = _normalize_optional_str(candidate.speech_id)
+            if candidate.llm_calls or candidate.tts_calls:
+                continue
+            if candidate.assistant_text.strip() or candidate.response_text.strip():
+                continue
+            if candidate.tool_step_announced or candidate.tool_executions:
+                continue
+            if candidate.last_tool_event_order is not None:
+                continue
+            if candidate.interrupted:
+                continue
+            if candidate_speech_id not in {None, anchor_speech_id}:
+                continue
+            if candidate.user_transcript.strip():
+                self._update_user_turn_text(
+                    anchor_turn,
+                    candidate.user_transcript,
+                    event_created_at=candidate.user_transcript_updated_at,
+                )
+            anchor_turn.user_turn_committed = (
+                anchor_turn.user_turn_committed or candidate.user_turn_committed
+            )
+            if candidate.user_turn_committed_at is not None:
+                anchor_turn.user_turn_committed_at = max(
+                    anchor_turn.user_turn_committed_at or candidate.user_turn_committed_at,
+                    candidate.user_turn_committed_at,
+                )
+            absorbed_turns.append(candidate)
+
+        for absorbed_turn in absorbed_turns:
+            self._pending_trace_turns.remove(absorbed_turn)
+            self._cancel_finalize_timeout(absorbed_turn.turn_id)
+
     def _select_turn_for_llm(self, speech_id: Optional[str]) -> Optional[TraceTurn]:
         if speech_id:
             matched = self._latest_turn_where(lambda c: c.speech_id == speech_id)
             if matched:
+                self._absorb_pending_pre_llm_turns(matched)
                 return matched
-            return self._next_turn_where(
+            matched = self._latest_turn_where(
                 lambda c: c.speech_id is None and not c.llm_calls
             )
-        return self._next_turn_where(lambda c: not c.llm_calls)
+            if matched is not None:
+                matched.speech_id = speech_id
+                self._absorb_pending_pre_llm_turns(matched)
+                return matched
+        matched = self._next_turn_where(lambda c: not c.llm_calls)
+        if matched is not None:
+            self._absorb_pending_pre_llm_turns(matched)
+        return matched
 
     def _select_turn_for_tts(self, speech_id: Optional[str]) -> Optional[TraceTurn]:
         if speech_id:
@@ -933,14 +1091,18 @@ class TurnTracer:
         *,
         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]
+        candidates = self._assistant_text_correlation_candidates()
+        if len(candidates) == 1:
+            turn = candidates[0]
+        else:
+            emit_ready_candidates = [
+                turn
+                for turn in candidates
+                if self._is_emit_ready(turn) and self._emit_ready_turn_is_recent(turn)
+            ]
+            if len(emit_ready_candidates) != 1:
+                return None
+            turn = emit_ready_candidates[0]
         cutoff = turn.orphan_assistant_cutoff_at
         if cutoff is not None:
             if event_created_at is None:
@@ -949,17 +1111,26 @@ class TurnTracer:
                 return None
         return turn
 
+    def _emit_ready_turn_is_recent(self, turn: TraceTurn) -> bool:
+        if turn.emit_ready_at is None:
+            return False
+        recent_window_sec = max(self._trace_finalize_timeout_sec, 1.0)
+        return (time() - turn.emit_ready_at) <= recent_window_sec
+
     def _apply_assistant_text_to_turn(
         self,
         turn: TraceTurn,
         assistant_text: str,
         *,
         event_created_at: Optional[float],
+        source: Optional[str],
     ) -> 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_missing = False
+        turn.assistant_text_source = source or turn.assistant_text_source or "unknown"
         turn.assistant_text_updated_at = _resolved_event_timestamp(event_created_at)
         turn.assistant_text_updated_order = order
         _reconcile_assistant_text_with_tts_calls(
@@ -975,6 +1146,7 @@ class TurnTracer:
         *,
         event_created_at: Optional[float],
         speech_id: Optional[str] = None,
+        source: Optional[str] = None,
     ) -> None:
         normalized = assistant_text.strip()
         if not normalized:
@@ -982,6 +1154,7 @@ class TurnTracer:
         record = AssistantTextRecord(
             text=normalized,
             event_created_at=_to_optional_float(event_created_at),
+            source=source,
         )
         normalized_speech_id = _normalize_optional_str(speech_id)
         if normalized_speech_id:
@@ -1010,6 +1183,7 @@ class TurnTracer:
                 turn,
                 record.text,
                 event_created_at=record.event_created_at,
+                source=record.source or "buffered_exact",
             )
         if not queue:
             self._pending_agent_transcripts_by_speech_id.pop(speech_id, None)
@@ -1035,6 +1209,7 @@ class TurnTracer:
                 turn,
                 record.text,
                 event_created_at=record.event_created_at,
+                source=record.source or "orphan_buffer",
             )
             return True
         return False
@@ -1072,12 +1247,7 @@ class TurnTracer:
         turn.perceived_latency_second_audio_ms = fallback_ms
 
     def _is_complete(self, turn: TraceTurn) -> bool:
-        base_complete = bool(
-            turn.user_transcript
-            and turn.assistant_text
-            and turn.llm_calls
-            and turn.tts_calls
-        )
+        base_complete = bool(self._is_emit_ready(turn) and turn.assistant_text)
         if not base_complete:
             return False
         if turn.tool_phase_open:
@@ -1086,8 +1256,7 @@ class TurnTracer:
 
     def _should_schedule_finalize_timeout(self, turn: TraceTurn) -> bool:
         return bool(
-            turn.llm_calls
-            and turn.tts_calls
+            self._is_emit_ready(turn)
             and not self._is_complete(turn)
             and not (turn.tool_phase_open and turn.last_tool_event_at is None)
             and self._resolve_finalize_timeout_sec(turn) > 0.0
@@ -1098,6 +1267,11 @@ class TurnTracer:
             return self._trace_post_tool_response_timeout_sec
         return self._trace_finalize_timeout_sec
 
+    def _finalize_wait_reason(self, turn: TraceTurn) -> str:
+        if self._requires_post_tool_response(turn):
+            return "post_tool_response"
+        return "assistant_text_grace"
+
     def _requires_post_tool_response(self, turn: TraceTurn) -> bool:
         if not turn.tool_step_announced and turn.last_tool_event_order is None:
             return False
@@ -1156,6 +1330,7 @@ class TurnTracer:
         missing_assistant_fallback: bool = False,
         tool_post_response_missing: bool = False,
         drop_assistant_text: bool = False,
+        finalization_reason: Optional[str] = None,
     ) -> TraceTurn:
         if drop_assistant_text:
             turn.assistant_text = ""
@@ -1169,36 +1344,62 @@ class TurnTracer:
             turn.assistant_text = turn.response_text
 
         if missing_assistant_fallback and not turn.assistant_text:
-            fallback = self._best_available_assistant_text(
+            fallback_text, fallback_source = self._best_available_assistant_text(
                 turn,
                 min_observed_order=(
                     turn.last_tool_event_order if tool_post_response_missing else None
                 ),
                 include_pending_agent_transcripts=not tool_post_response_missing,
             )
-            if fallback:
-                turn.assistant_text = fallback
-                if not turn.response_text:
-                    turn.response_text = fallback
+            if fallback_text:
+                self._apply_assistant_text_to_turn(
+                    turn,
+                    fallback_text,
+                    event_created_at=None,
+                    source=fallback_source or "unknown",
+                )
             else:
                 tool_error_fallback = ""
                 if tool_post_response_missing:
                     tool_error_fallback = _tool_error_fallback_text(turn)
 
                 if tool_error_fallback:
-                    turn.assistant_text = tool_error_fallback
-                    if not turn.response_text:
-                        turn.response_text = tool_error_fallback
+                    self._apply_assistant_text_to_turn(
+                        turn,
+                        tool_error_fallback,
+                        event_created_at=None,
+                        source="tool_fallback",
+                    )
                 else:
                     turn.assistant_text_missing = True
                     unavailable = "[assistant text unavailable]"
                     turn.assistant_text = unavailable
+                    turn.assistant_text_source = "unavailable"
                     if not turn.response_text:
                         turn.response_text = unavailable
+                    logger.warning(
+                        "Langfuse turn finalized without assistant text: turn_id=%s speech_id=%s reason=%s",
+                        turn.turn_id,
+                        turn.speech_id,
+                        finalization_reason
+                        or ("post_tool_timeout" if tool_post_response_missing else "assistant_text_grace_timeout"),
+                    )
 
         turn.tool_phase_open = False
         if tool_post_response_missing:
             turn.tool_post_response_missing = True
+        if finalization_reason is None:
+            if turn.interrupted_reason == "shutdown_drain":
+                finalization_reason = "shutdown_drain"
+            elif turn.interrupted and turn.assistant_audio_started:
+                finalization_reason = "interrupted_after_audio"
+            elif tool_post_response_missing:
+                finalization_reason = "post_tool_timeout"
+            elif missing_assistant_fallback:
+                finalization_reason = "assistant_text_grace_timeout"
+            else:
+                finalization_reason = "complete"
+        turn.finalization_reason = finalization_reason
 
         self._pending_trace_turns.remove(turn)
         self._cancel_finalize_timeout(turn.turn_id)
@@ -1210,7 +1411,7 @@ class TurnTracer:
         *,
         min_observed_order: Optional[int] = None,
         include_pending_agent_transcripts: bool = True,
-    ) -> str:
+    ) -> tuple[str, Optional[str]]:
         speech_id = _normalize_optional_str(turn.speech_id)
         if turn.assistant_text.strip():
             if (
@@ -1220,7 +1421,7 @@ class TurnTracer:
                     and turn.assistant_text_updated_order > min_observed_order
                 )
             ):
-                return turn.assistant_text.strip()
+                return turn.assistant_text.strip(), turn.assistant_text_source
         if turn.response_text.strip():
             if (
                 min_observed_order is None
@@ -1229,7 +1430,7 @@ class TurnTracer:
                     and turn.assistant_text_updated_order > min_observed_order
                 )
             ):
-                return turn.response_text.strip()
+                return turn.response_text.strip(), turn.assistant_text_source
         if speech_id:
             buffered_exact = self._pending_agent_transcripts_by_speech_id.get(speech_id)
             if buffered_exact:
@@ -1239,7 +1440,7 @@ class TurnTracer:
                         continue
                     if not buffered_exact:
                         self._pending_agent_transcripts_by_speech_id.pop(speech_id, None)
-                    return record.text.strip()
+                    return record.text.strip(), record.source or "buffered_exact"
         for tts_call in reversed(turn.tts_calls):
             if (
                 min_observed_order is not None
@@ -1247,16 +1448,39 @@ class TurnTracer:
             ):
                 continue
             if tts_call.assistant_text.strip():
-                return tts_call.assistant_text.strip()
+                return tts_call.assistant_text.strip(), "tts_metrics"
         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 ""
+                return turn.assistant_text.strip(), turn.assistant_text_source
+        if (
+            include_pending_agent_transcripts
+            and self._pending_agent_transcripts
+            and self._can_consume_pending_agent_transcript_for_turn(turn)
+        ):
+            return self._pending_agent_transcripts.popleft().strip(), "pending_agent_transcript"
+        return "", None
+
+    def _assistant_text_correlation_candidates(self) -> list[TraceTurn]:
+        return [
+            turn
+            for turn in self._pending_trace_turns
+            if bool(turn.llm_calls or turn.tts_calls or turn.tool_phase_open)
+            and not (turn.interrupted and turn.assistant_audio_started)
+        ]
+
+    def _can_consume_pending_agent_transcript_for_turn(self, turn: TraceTurn) -> bool:
+        candidates = self._assistant_text_correlation_candidates()
+        return len(candidates) == 1 and candidates[0] is turn
 
-    def _finalize_interrupted_turns_before_new_user_turn_locked(self) -> list[TraceTurn]:
+    def _has_active_finalize_timeout(self, turn_id: str) -> bool:
+        task = self._trace_finalize_tasks.get(turn_id)
+        return task is not None and not task.done()
+
+    def _finalize_interrupted_turns_before_new_user_turn_locked(
+        self,
+    ) -> tuple[list[TraceTurn], list[tuple[str, float]]]:
         completed_turns: list[TraceTurn] = []
+        timeout_schedules: list[tuple[str, float]] = []
         for turn in list(self._pending_trace_turns):
             if not (turn.user_transcript and turn.llm_calls and turn.tts_calls):
                 continue
@@ -1269,17 +1493,49 @@ class TurnTracer:
             missing_post_tool_assistant = bool(
                 requires_post_tool_response and not self._post_tool_assistant_observed(turn)
             )
-            completed_turns.append(
-                self._finalize_locked(
+            if not turn.assistant_text.strip() or missing_post_tool_assistant:
+                fallback_text, fallback_source = self._best_available_assistant_text(
                     turn,
-                    missing_assistant_fallback=(
-                        missing_post_tool_assistant or not bool(turn.assistant_text)
+                    min_observed_order=(
+                        turn.last_tool_event_order if missing_post_tool_assistant else None
                     ),
-                    tool_post_response_missing=requires_post_tool_response,
-                    drop_assistant_text=missing_post_tool_assistant,
+                    include_pending_agent_transcripts=not missing_post_tool_assistant,
                 )
-            )
-        return completed_turns
+                if fallback_text:
+                    self._apply_assistant_text_to_turn(
+                        turn,
+                        fallback_text,
+                        event_created_at=None,
+                        source=fallback_source or "unknown",
+                    )
+                    missing_post_tool_assistant = False
+
+            if turn.assistant_text.strip() and not missing_post_tool_assistant:
+                completed_turns.append(
+                    self._finalize_locked(
+                        turn,
+                        missing_assistant_fallback=False,
+                        tool_post_response_missing=False,
+                        drop_assistant_text=False,
+                    )
+                )
+                continue
+
+            timeout_sec = self._resolve_finalize_timeout_sec(turn)
+            if timeout_sec <= 0.0:
+                completed_turns.append(
+                    self._finalize_locked(
+                        turn,
+                        missing_assistant_fallback=True,
+                        tool_post_response_missing=requires_post_tool_response,
+                        drop_assistant_text=missing_post_tool_assistant,
+                    )
+                )
+                continue
+            if self._has_active_finalize_timeout(turn.turn_id):
+                continue
+            timeout_schedules.append((turn.turn_id, timeout_sec))
+        return completed_turns, timeout_schedules
 
     def _requires_post_tool_follow_up(self, turn: TraceTurn) -> bool:
         if turn.last_tool_event_order is None:
@@ -1310,6 +1566,15 @@ class TurnTracer:
             )
         )
         self._trace_finalize_tasks[turn_id] = task
+        turn = next((t for t in self._pending_trace_turns if t.turn_id == turn_id), None)
+        if turn is not None:
+            logger.debug(
+                "Scheduled Langfuse finalize wait: turn_id=%s speech_id=%s timeout_sec=%.3f wait_reason=%s",
+                turn.turn_id,
+                turn.speech_id,
+                timeout_sec,
+                self._finalize_wait_reason(turn),
+            )
         task.add_done_callback(
             lambda _task, tid=turn_id, v=version: self._on_finalize_timeout_task_done(
                 turn_id=tid,
@@ -1549,12 +1814,17 @@ class TurnTracer:
                             for llm_idx, llm_call in enumerate(
                                 block.llm_calls, start=1
                             ):
+                                llm_visible_latency_ms = _preferred_visible_latency_ms(
+                                    llm_call.ttft_ms,
+                                    llm_call.duration_ms,
+                                )
                                 phase_cursor_ns = _emit_component_span(
                                     _tracer,
                                     name="LLMMetrics",
                                     context=phase_ctx,
                                     start_ns=phase_cursor_ns,
-                                    duration_ms=llm_call.duration_ms,
+                                    duration_ms=llm_visible_latency_ms,
+                                    advance_ms=llm_call.duration_ms,
                                     attributes=_merge_component_attributes(
                                         llm_call.attributes,
                                         {
@@ -1562,6 +1832,7 @@ class TurnTracer:
                                             "response_text": phase_text,
                                             "ttft_ms": llm_call.ttft_ms,
                                             "llm_total_latency_ms": llm_call.duration_ms,
+                                            "total_duration_ms": llm_call.duration_ms,
                                             "phase_index": block.index,
                                             "phase_call_index": llm_idx,
                                         },
@@ -1574,18 +1845,25 @@ class TurnTracer:
                                 block.tts_calls, start=1
                             ):
                                 spoken_text = tts_call.assistant_text or phase_text
+                                tts_visible_latency_ms = _preferred_visible_latency_ms(
+                                    tts_call.ttfb_ms,
+                                    tts_call.duration_ms,
+                                )
                                 phase_cursor_ns = _emit_component_span(
                                     _tracer,
                                     name="TTSMetrics",
                                     context=phase_ctx,
                                     start_ns=phase_cursor_ns,
-                                    duration_ms=tts_call.duration_ms,
+                                    duration_ms=tts_visible_latency_ms,
+                                    advance_ms=tts_call.duration_ms,
                                     attributes=_merge_component_attributes(
                                         tts_call.attributes,
                                         {
                                             "assistant_text": spoken_text,
                                             "assistant_text_missing": turn.assistant_text_missing,
                                             "ttfb_ms": tts_call.ttfb_ms,
+                                            "tts_total_latency_ms": tts_call.duration_ms,
+                                            "total_duration_ms": tts_call.duration_ms,
                                             "phase_index": block.index,
                                             "phase_call_index": tts_idx,
                                         },
@@ -1699,12 +1977,17 @@ class TurnTracer:
                 )
 
             logger.info(
-                "Langfuse turn trace emitted: trace_id=%s turn_id=%s session_id=%s room_id=%s participant_id=%s",
+                "Langfuse turn trace emitted: trace_id=%s turn_id=%s session_id=%s room_id=%s participant_id=%s finalization_reason=%s assistant_text_source=%s emit_wait_ms=%.1f",
                 turn.trace_id,
                 turn.turn_id,
                 turn.session_id,
                 turn.room_id,
                 turn.participant_id,
+                turn.finalization_reason,
+                turn.assistant_text_source,
+                max((time() - turn.emit_ready_at) * 1000.0, 0.0)
+                if turn.emit_ready_at is not None
+                else 0.0,
             )
             asyncio.create_task(self._flush_tracer_provider())
         except Exception as exc:
@@ -2422,6 +2705,7 @@ def _set_root_attributes(
         "langfuse.trace.metadata.participant_id": turn.participant_id,
         "langfuse.trace.metadata.turn_id": turn.turn_id,
         "langfuse.trace.metadata.assistant_text_missing": turn.assistant_text_missing,
+        "langfuse.trace.metadata.assistant_text_source": turn.assistant_text_source,
         "langfuse.trace.metadata.stt_status": turn.stt_status,
         "langfuse.trace.metadata.tool_phase_announced": turn.tool_step_announced,
         "langfuse.trace.metadata.tool_post_response_missing": turn.tool_post_response_missing,
@@ -2429,6 +2713,8 @@ def _set_root_attributes(
         "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.finalization_reason": turn.finalization_reason,
+        "langfuse.trace.metadata.emit_ready_at": turn.emit_ready_at,
         "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,
@@ -2510,6 +2796,30 @@ def _duration_attribute_to_ms(value: Any) -> Optional[float]:
     return None
 
 
+def _assistant_text_from_component_attributes(attributes: dict[str, Any]) -> str:
+    for key in (
+        "assistant_text",
+        "spoken_text",
+        "metadata.assistant_text",
+        "metadata.spoken_text",
+    ):
+        value = attributes.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return ""
+
+
+def _preferred_visible_latency_ms(
+    preferred_ms: Optional[float],
+    fallback_ms: Optional[float],
+) -> Optional[float]:
+    if preferred_ms is not None and preferred_ms >= 0.0:
+        return preferred_ms
+    if fallback_ms is not None and fallback_ms >= 0.0:
+        return fallback_ms
+    return None
+
+
 def _tool_calls_total_duration_ms(tool_calls: list[ToolCallTrace]) -> float:
     total = 0.0
     for call in tool_calls:
@@ -2538,6 +2848,10 @@ def _merge_user_transcripts(existing: str, incoming: str) -> str:
         return left
     if left == right:
         return left
+    if left.casefold() in right.casefold():
+        return right
+    if right.casefold() in left.casefold():
+        return left
     if right.startswith(left):
         return right
     if left.startswith(right):
@@ -2562,12 +2876,19 @@ def _emit_component_span(
     context: Any,
     start_ns: int,
     duration_ms: Optional[float],
+    advance_ms: Optional[float] = None,
     attributes: dict[str, Any],
     observation_input: Optional[str] = None,
     observation_output: Optional[str] = None,
 ) -> int:
     actual_ms = max(duration_ms, 0.0) if duration_ms is not None else None
+    cursor_advance_ms = actual_ms
+    if advance_ms is not None:
+        cursor_advance_ms = max(advance_ms, 0.0)
+        if actual_ms is not None:
+            cursor_advance_ms = max(cursor_advance_ms, actual_ms)
     end_ns = start_ns + _ms_to_ns(actual_ms or 0.0)
+    next_cursor_ns = start_ns + _ms_to_ns(cursor_advance_ms or 0.0)
 
     span = _tracer.start_span(name, context=context, start_time=start_ns)
     try:
@@ -2584,7 +2905,7 @@ def _emit_component_span(
                 span.set_attribute(key, value)
     finally:
         _close_span_at(span, end_ns)
-    return end_ns
+    return max(end_ns, next_cursor_ns)
 
 
 def _close_span_at(span: Any, end_ns: int) -> None:

src/core/settings.py

@@ -198,7 +198,7 @@ class VoiceSettings(CoreSettings):
 class STTSettings(CoreSettings):
     # Provider selection
     STT_PROVIDER: str = Field(
-        default="deepgram",
+        default="moonshine",
         description="STT provider: 'moonshine', 'nvidia', or 'deepgram'"
     )
 
@@ -395,11 +395,23 @@ class LangfuseSettings(CoreSettings):
         default=True,
         description="Mark emitted Langfuse traces as public for shareable URLs",
     )
+    LANGFUSE_ASSISTANT_TEXT_GRACE_TIMEOUT_MS: float = Field(
+        default=500.0,
+        ge=0.0,
+        le=10000.0,
+        description=(
+            "Short grace window to wait for assistant text on regular turns before "
+            "force-finalizing the trace"
+        ),
+    )
     LANGFUSE_TRACE_FINALIZE_TIMEOUT_MS: float = Field(
         default=8000.0,
         ge=0.0,
         le=10000.0,
-        description="Timeout to wait for assistant text before force-finalizing trace",
+        description=(
+            "Legacy finalize timeout retained for compatibility; regular turns use "
+            "LANGFUSE_ASSISTANT_TEXT_GRACE_TIMEOUT_MS"
+        ),
     )
     LANGFUSE_POST_TOOL_RESPONSE_TIMEOUT_MS: float = Field(
         default=30000.0,

src/plugins/pocket_tts/tts.py

@@ -8,7 +8,7 @@ import threading
 import time
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
-from typing import Any, Protocol, cast
+from typing import Any, cast
 
 import numpy as np
 from pocket_tts import TTSModel
@@ -35,14 +35,6 @@ _SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?])\s+")
 _WHITESPACE_RE = re.compile(r"\s+")
 
 
-class TTSMetricsCallback(Protocol):
-    # duration is end-to-end wall-clock synth time, not model-only compute time.
-    def __call__(self, *, ttfb: float, duration: float, audio_duration: float) -> None: ...
-
-
-OptionalTTSMetricsCallback = TTSMetricsCallback | None
-
-
 @dataclass
 class _GenerationError:
     error: Exception
@@ -61,7 +53,6 @@ class PocketTTS(tts.TTS):
         lsd_decode_steps: int = 1,
         sample_rate: int = NATIVE_SAMPLE_RATE,
         max_concurrent_generations: int = 1,
-        metrics_callback: OptionalTTSMetricsCallback = None,
     ) -> None:
         """Create a new instance of Pocket TTS.
 
@@ -72,7 +63,6 @@ class PocketTTS(tts.TTS):
             sample_rate: Output sample rate. Only native 24kHz is supported.
             max_concurrent_generations: Maximum number of concurrent synthesis tasks
                 for this PocketTTS instance.
-            metrics_callback: Optional callback for per-segment generation metrics.
         """
         if max_concurrent_generations < 1:
             raise ValueError(
@@ -95,8 +85,6 @@ class PocketTTS(tts.TTS):
         self._temperature = temperature
         self._lsd_decode_steps = lsd_decode_steps
         self._max_concurrent_generations = max_concurrent_generations
-        self._metrics_callback = metrics_callback
-
         self._model: Any = TTSModel.load_model(temp=temperature, lsd_decode_steps=lsd_decode_steps)
         self._voice_state: Any = self._load_voice_state(voice)
         self._generation_semaphore = asyncio.Semaphore(max_concurrent_generations)
@@ -336,13 +324,6 @@ class PocketChunkedStream(tts.ChunkedStream):
 
         output_emitter.flush()
 
-        if pocket_tts._metrics_callback and first_chunk_ttfb >= 0:
-            pocket_tts._metrics_callback(
-                ttfb=first_chunk_ttfb,
-                duration=total_synth_wall_time,
-                audio_duration=audio_duration,
-            )
-
 
 class PocketSynthesizeStream(tts.SynthesizeStream):
     def __init__(self, *, tts: PocketTTS, conn_options: APIConnectOptions) -> None:
@@ -398,13 +379,6 @@ class PocketSynthesizeStream(tts.SynthesizeStream):
         finally:
             output_emitter.end_segment()
 
-        if pocket_tts._metrics_callback and first_chunk_ttfb >= 0:
-            pocket_tts._metrics_callback(
-                ttfb=first_chunk_ttfb,
-                duration=total_synth_wall_time,
-                audio_duration=audio_duration,
-            )
-
     async def _synthesize_segment(
         self, text: str, output_emitter: tts.AudioEmitter
     ) -> tuple[float, float, float]:

tests/test_langfuse_turn_tracing.py

@@ -161,6 +161,7 @@ def _make_tts_metrics(
     ttfb: float = 0.15,
     duration: float = 0.5,
     audio_duration: float = 1.3,
+    metadata: Any = None,
 ) -> metrics.TTSMetrics:
     return metrics.TTSMetrics(
         label="tts",
@@ -173,6 +174,7 @@ def _make_tts_metrics(
         characters_count=42,
         streamed=True,
         speech_id=speech_id,
+        metadata=metadata,
     )
 
 
@@ -244,6 +246,34 @@ class _FakeTextMethodPart:
 
 
 class _FakeSpeechHandle:
+    def __init__(self, chat_items: list[Any], speech_id: str = "speech-fake") -> None:
+        self.id = speech_id
+        self.chat_items = chat_items
+        self._callbacks: list[Any] = []
+        self._item_added_callbacks: list[Any] = []
+
+    def add_done_callback(self, callback: Any) -> None:
+        self._callbacks.append(callback)
+
+    def _add_item_added_callback(self, callback: Any) -> None:
+        self._item_added_callbacks.append(callback)
+
+    def _remove_item_added_callback(self, callback: Any) -> None:
+        self._item_added_callbacks = [
+            registered for registered in self._item_added_callbacks if registered is not callback
+        ]
+
+    def add_chat_item(self, item: Any) -> None:
+        self.chat_items.append(item)
+        for callback in list(self._item_added_callbacks):
+            callback(item)
+
+    def trigger_done(self) -> None:
+        for callback in self._callbacks:
+            callback(self)
+
+
+class _FakeSpeechHandleWithoutItemAddedHook:
     def __init__(self, chat_items: list[Any], speech_id: str = "speech-fake") -> None:
         self.id = speech_id
         self.chat_items = chat_items
@@ -346,6 +376,8 @@ def test_turn_trace_has_required_metadata_and_spans(monkeypatch: pytest.MonkeyPa
     assert root.attributes["turn_id"]
     assert root.attributes["langfuse.trace.output"] == "hi, how can I help?"
     assert root.attributes["langfuse.trace.public"] is False
+    assert root.attributes["langfuse.trace.metadata.finalization_reason"] == "complete"
+    assert root.attributes["langfuse.trace.metadata.assistant_text_source"] == "conversation_item"
     assert root.attributes["latency_ms.eou_delay"] == pytest.approx(1100.0)
     assert root.attributes["latency_ms.stt_finalization"] == pytest.approx(250.0)
     assert root.attributes["latency_ms.stt_total"] == pytest.approx(1350.0)
@@ -377,17 +409,28 @@ def test_turn_trace_has_required_metadata_and_spans(monkeypatch: pytest.MonkeyPa
     assert llm_span.attributes["response_text"] == "hi, how can I help?"
     assert llm_span.attributes["ttft_ms"] > 0
     assert llm_span.attributes["llm_total_latency_ms"] > 0
+    assert llm_span.attributes["total_duration_ms"] == pytest.approx(
+        llm_span.attributes["llm_total_latency_ms"]
+    )
     assert llm_span.attributes["input"] == "hello there"
     assert llm_span.attributes["output"] == "hi, how can I help?"
-    assert llm_span.attributes["duration_ms"] > 0
+    assert llm_span.attributes["duration_ms"] == pytest.approx(
+        llm_span.attributes["ttft_ms"]
+    )
     assert llm_span.attributes["prompt_tokens"] == 12
     assert llm_span.attributes["completion_tokens"] == 24
 
     assert tts_span.attributes["assistant_text"] == "hi, how can I help?"
     assert tts_span.attributes["ttfb_ms"] > 0
+    assert tts_span.attributes["tts_total_latency_ms"] > 0
+    assert tts_span.attributes["total_duration_ms"] == pytest.approx(
+        tts_span.attributes["tts_total_latency_ms"]
+    )
     assert tts_span.attributes["input"] == "hi, how can I help?"
     assert tts_span.attributes["output"] == "hi, how can I help?"
-    assert tts_span.attributes["duration_ms"] > 0
+    assert tts_span.attributes["duration_ms"] == pytest.approx(
+        tts_span.attributes["ttfb_ms"]
+    )
     assert tts_span.attributes["characters_count"] == 42
     assert tts_span.attributes["streamed"] is True
 
@@ -1734,6 +1777,152 @@ def test_multiple_final_transcripts_are_merged_into_one_turn(
     assert root.attributes["langfuse.trace.metadata.coalesced_turn_count"] == 0
 
 
+def test_same_speech_final_transcripts_keep_merging_after_eou_until_llm(
+    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-same-speech-final-merge-after-eou",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("Hello there.", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-final-merge-after-eou", delay=0.9, transcription_delay=0.2)
+        )
+        await collector.on_user_input_transcribed("I'm missing context.", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-final-merge-after-eou"))
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="Hi there.",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-final-merge-after-eou"))
+        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"] == "Hello there. I'm missing context."
+    assert root.attributes["langfuse.trace.output"] == "Hi there."
+    assert root.attributes["langfuse.trace.metadata.assistant_text_missing"] is False
+
+
+def test_user_conversation_item_after_eou_merges_instead_of_replacing(
+    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-user-item-merges-after-eou",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("Hello there.", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-user-item-after-eou", delay=0.8, transcription_delay=0.2)
+        )
+        await collector.on_conversation_item_added(
+            role="user",
+            content="I'm missing the rest.",
+        )
+        await collector.on_metrics_collected(_make_llm_metrics("speech-user-item-after-eou"))
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="Thanks for clarifying.",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-user-item-after-eou"))
+        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"] == "Hello there. I'm missing the rest."
+    assert root.attributes["langfuse.trace.output"] == "Thanks for clarifying."
+
+
+def test_same_speech_fragmented_input_with_late_speech_done_keeps_full_trace(
+    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,
+    )
+    handle = _FakeSpeechHandle(chat_items=[], speech_id="speech-fragmented-late-done")
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-fragmented-late-done",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("Hello there.", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-fragmented-late-done", delay=0.8, transcription_delay=0.2)
+        )
+        await collector.on_user_input_transcribed("I'm missing context.", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-fragmented-late-done"))
+        await collector.on_speech_created(handle)
+        await collector.on_metrics_collected(_make_tts_metrics("speech-fragmented-late-done"))
+        await collector.wait_for_pending_trace_tasks()
+
+        turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+        assert not turn_spans
+
+        handle.chat_items = [_FakeChatItem(role="assistant", content=["Hi there."])]
+        handle.trigger_done()
+        await asyncio.sleep(0)
+        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"] == "Hello there. I'm missing context."
+    assert root.attributes["langfuse.trace.output"] == "Hi there."
+    assert root.attributes["langfuse.trace.metadata.assistant_text_missing"] is False
+
+
 def test_immediate_continuation_coalesces_aborted_prior_turn(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -2089,6 +2278,103 @@ def test_trace_waits_for_assistant_text_before_emit(monkeypatch: pytest.MonkeyPa
     assert turn_spans[0].attributes["langfuse.trace.output"] == "hello there"
 
 
+def test_tts_metric_metadata_assistant_text_emits_without_placeholder(
+    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-tts-metadata",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("hi", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-tts-metadata"))
+        tts_metrics = _make_tts_metrics("speech-tts-metadata")
+        tts_metrics.metadata = {
+            "model_name": "pocket-tts",
+            "model_provider": "Kyutai",
+            "assistant_text": "assistant text from tts metadata",
+        }
+        await collector.on_metrics_collected(tts_metrics)
+        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.output"] == "assistant text from tts metadata"
+    assert root.attributes["langfuse.trace.metadata.assistant_text_missing"] is False
+    assert root.attributes["langfuse.trace.metadata.assistant_text_source"] == "tts_metrics"
+    assert root.attributes["langfuse.trace.metadata.finalization_reason"] == "complete"
+
+
+def test_speech_item_added_assistant_text_arriving_within_grace_emits_trace(
+    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,
+    )
+    collector._trace_finalize_timeout_sec = 0.05
+    handle = _FakeSpeechHandle([], speech_id="speech-item-added")
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-speech-item-added",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("hi", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-item-added"))
+        await collector.on_speech_created(handle)
+        await collector.on_metrics_collected(_make_tts_metrics("speech-item-added"))
+        await collector.wait_for_pending_trace_tasks()
+        assert not fake_tracer.spans
+
+        handle.add_chat_item(
+            _FakeChatItem(
+                role="assistant",
+                content=["assistant text from speech item callback"],
+            )
+        )
+        await asyncio.sleep(0)
+        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.output"] == "assistant text from speech item callback"
+    assert root.attributes["langfuse.trace.metadata.assistant_text_missing"] is False
+    assert root.attributes["langfuse.trace.metadata.assistant_text_source"] == "speech_item_added"
+    assert root.attributes["langfuse.trace.metadata.finalization_reason"] == "complete"
+
+
 def test_speech_created_done_callback_backfills_assistant_text(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -2118,7 +2404,7 @@ def test_speech_created_done_callback_backfills_assistant_text(
         await collector.wait_for_pending_trace_tasks()
         assert not fake_tracer.spans
 
-        handle = _FakeSpeechHandle(
+        handle = _FakeSpeechHandleWithoutItemAddedHook(
             chat_items=[_FakeChatItem(role="assistant", content=["fallback reply"])],
             speech_id="speech-speech-created",
         )
@@ -2135,6 +2421,56 @@ def test_speech_created_done_callback_backfills_assistant_text(
     assert turn_spans[0].attributes["langfuse.trace.metadata.assistant_text_missing"] is False
 
 
+def test_speech_done_does_not_replace_speech_item_added_text(
+    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,
+    )
+    handle = _FakeSpeechHandle([], speech_id="speech-speech-item-preferred")
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-speech-item-preferred",
+            participant_id="web-123",
+        )
+        await collector.on_user_input_transcribed("hi there", is_final=True)
+        await collector.on_metrics_collected(_make_llm_metrics("speech-speech-item-preferred"))
+        await collector.on_speech_created(handle)
+        handle.add_chat_item(
+            _FakeChatItem(role="assistant", content=["preferred reply from speech item"])
+        )
+        await asyncio.sleep(0)
+
+        handle.chat_items.append(
+            _FakeChatItem(role="assistant", content=["stale reply from speech done"])
+        )
+        handle.trigger_done()
+        await asyncio.sleep(0)
+
+        await collector.on_metrics_collected(_make_tts_metrics("speech-speech-item-preferred"))
+        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.output"] == "preferred reply from speech item"
+    assert root.attributes["langfuse.trace.metadata.assistant_text_source"] == "speech_item_added"
+
+
 def test_speech_created_immediate_capture_backfills_assistant_text(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -2221,6 +2557,8 @@ def test_trace_finalize_timeout_for_missing_assistant_text(
     root = turn_spans[0]
     assert root.attributes["langfuse.trace.metadata.assistant_text_missing"] is True
     assert root.attributes["langfuse.trace.output"] == "[assistant text unavailable]"
+    assert root.attributes["langfuse.trace.metadata.assistant_text_source"] == "unavailable"
+    assert root.attributes["langfuse.trace.metadata.finalization_reason"] == "assistant_text_grace_timeout"
 
 
 def test_trace_finalize_timeout_uses_pending_assistant_transcript(
@@ -2570,7 +2908,7 @@ def test_stale_orphan_assistant_text_from_absorbed_turn_is_not_attached_to_conti
     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(
+def test_audio_started_turn_waits_for_timeout_before_placeholder_on_barge_in(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     import src.agent.traces.metrics_collector as metrics_collector_module
@@ -2587,6 +2925,7 @@ def test_audio_started_turn_is_finalized_separately_when_new_user_turn_arrives(
         participant_id="web-123",
         langfuse_enabled=True,
     )
+    collector._trace_finalize_timeout_sec = 0.01
 
     async def _run() -> None:
         await collector.on_session_metadata(
@@ -2619,15 +2958,116 @@ def test_audio_started_turn_is_finalized_separately_when_new_user_turn_arrives(
         await collector.on_metrics_collected(_make_tts_metrics("speech-separate-b"))
         await collector.wait_for_pending_trace_tasks()
 
+        turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+        assert len(turn_spans) == 1
+        assert turn_spans[0].attributes["langfuse.trace.input"] == "second prompt"
+
+        await asyncio.sleep(0.03)
+        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"
+    first = next(
+        span
+        for span in turn_spans
+        if span.attributes["langfuse.trace.input"] == "first prompt"
+    )
+    second = next(
+        span
+        for span in turn_spans
+        if span.attributes["langfuse.trace.input"] == "second 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_audio_started_turn_uses_late_speech_done_text_after_barge_in(
+    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,
+    )
+    collector._trace_finalize_timeout_sec = 0.05
+    handle = _FakeSpeechHandle(chat_items=[], speech_id="speech-audio-started-late-done")
+
+    async def _run() -> None:
+        await collector.on_session_metadata(
+            session_id="session-audio-started-late-done",
+            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-late-done",
+                delay=0.4,
+                transcription_delay=0.1,
+            )
+        )
+        await collector.on_metrics_collected(
+            _make_llm_metrics("speech-audio-started-late-done")
+        )
+        await collector.on_speech_created(handle)
+        await collector.on_agent_state_changed(
+            old_state="thinking",
+            new_state="speaking",
+        )
+        await collector.on_metrics_collected(
+            _make_tts_metrics("speech-audio-started-late-done")
+        )
+
+        await collector.on_user_input_transcribed("second prompt", is_final=True)
+        await collector.on_metrics_collected(
+            _make_eou_metrics("speech-separate-late-done-b", delay=0.5, transcription_delay=0.1)
+        )
+        await collector.on_metrics_collected(_make_llm_metrics("speech-separate-late-done-b"))
+        await collector.on_conversation_item_added(
+            role="assistant",
+            content="second reply",
+        )
+        await collector.on_metrics_collected(_make_tts_metrics("speech-separate-late-done-b"))
+        await collector.wait_for_pending_trace_tasks()
+
+        turn_spans = [span for span in fake_tracer.spans if span.name == "turn"]
+        assert len(turn_spans) == 1
+        assert turn_spans[0].attributes["langfuse.trace.input"] == "second prompt"
+
+        handle.chat_items = [
+            _FakeChatItem(role="assistant", content=["first reply recovered late"])
+        ]
+        handle.trigger_done()
+        await asyncio.sleep(0)
+        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 = next(
+        span
+        for span in turn_spans
+        if span.attributes["langfuse.trace.input"] == "first prompt"
+    )
+    second = next(
+        span
+        for span in turn_spans
+        if span.attributes["langfuse.trace.input"] == "second prompt"
+    )
+    assert first.attributes["langfuse.trace.metadata.interrupted"] is True
+    assert first.attributes["langfuse.trace.output"] == "first reply recovered late"
     assert second.attributes["langfuse.trace.output"] == "second reply"