Fix realtime feedback event correlation (#534)

api/services/pipecat/event_handlers.py

@@ -14,6 +14,7 @@ from api.services.pipecat.in_memory_buffers import (
 )
 from api.services.pipecat.pipeline_metrics_aggregator import PipelineMetricsAggregator
 from api.services.pipecat.tracing_config import get_trace_url
+from api.services.pipecat.transcript_log_coordinator import TranscriptLogCoordinator
 from api.services.posthog_client import capture_event
 from api.services.workflow.pipecat_engine import PipecatEngine
 from api.services.workflow_run_artifacts import upload_workflow_run_artifacts
@@ -65,6 +66,7 @@ def register_event_handlers(
     engine: PipecatEngine,
     audio_buffer: AudioBufferProcessor,
     in_memory_logs_buffer: InMemoryLogsBuffer,
+    transcript_log_coordinator: TranscriptLogCoordinator,
     pipeline_metrics_aggregator: PipelineMetricsAggregator,
     audio_config=AudioConfig,
     pre_call_fetch_task: asyncio.Task | None = None,
@@ -223,7 +225,12 @@ def register_event_handlers(
         task: PipelineWorker,
         _frame: Frame,
     ):
-        logger.debug(f"In on_pipeline_finished callback handler")
+        logger.debug("In on_pipeline_finished callback handler")
+
+        # Turn and feedback observers run on independent queues. Drain them
+        # before finalizing immutable transcripts and taking the DB snapshot.
+        await task.wait_for_observers()
+        await transcript_log_coordinator.flush()
 
         workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id)
 

api/services/pipecat/in_memory_buffers.py

@@ -1,6 +1,7 @@
 import asyncio
 import io
 import wave
+from copy import deepcopy
 from datetime import UTC, datetime
 from typing import List, Optional
 
@@ -104,15 +105,9 @@ class InMemoryLogsBuffer:
     def __init__(self, workflow_run_id: int):
         self._workflow_run_id = workflow_run_id
         self._events: List[dict] = []
-        self._turn_counter = 0
+        self._current_turn: Optional[int] = None
         self._current_node_id: Optional[str] = None
         self._current_node_name: Optional[str] = None
-        self._user_speech_start_timestamp: Optional[str] = None
-        self._user_speech_end_timestamp: Optional[str] = None
-        self._user_speech_start_from_vad = False
-        self._user_speech_end_from_vad = False
-        self._bot_speech_start_timestamp: Optional[str] = None
-        self._bot_speech_end_timestamp: Optional[str] = None
 
     def set_current_node(self, node_id: str, node_name: str):
         """Set the current node ID and name to be injected into subsequent events."""
@@ -129,153 +124,44 @@ class InMemoryLogsBuffer:
         """Get the current node name."""
         return self._current_node_name
 
-    @staticmethod
-    def _now_iso() -> str:
-        return datetime.now(UTC).isoformat(timespec="milliseconds")
+    def set_current_turn(self, turn: int) -> None:
+        """Set the fallback turn for non-transcript events."""
+        self._current_turn = turn
 
-    def mark_user_started_speaking(
-        self, timestamp: Optional[str] = None, *, from_vad: bool = False
+    async def append(
+        self,
+        event: dict,
+        *,
+        timestamp: Optional[str] = None,
+        turn: Optional[int] = None,
+        node_id: Optional[str] = None,
+        node_name: Optional[str] = None,
+        use_current_node: bool = True,
     ):
-        """Record when the user started speaking for the current turn."""
-        vad_interval_is_open = (
-            self._user_speech_start_from_vad and self._user_speech_end_timestamp is None
-        )
-        if vad_interval_is_open and not from_vad:
-            return
-
-        self._user_speech_start_timestamp = timestamp or self._now_iso()
-        self._user_speech_end_timestamp = None
-        self._user_speech_start_from_vad = from_vad
-        self._user_speech_end_from_vad = False
-        self._update_latest_payload_start_timestamp(
-            RealtimeFeedbackType.USER_TRANSCRIPTION.value,
-            self._user_speech_start_timestamp,
-            require_final=True,
-        )
-
-    def mark_user_stopped_speaking(
-        self, timestamp: Optional[str] = None, *, from_vad: bool = False
-    ):
-        """Record when the user stopped speaking and update the latest user event."""
-        if self._user_speech_end_from_vad and not from_vad:
-            return
-
-        self._user_speech_end_timestamp = timestamp or self._now_iso()
-        self._user_speech_end_from_vad = from_vad
-        self._update_latest_payload_end_timestamp(
-            RealtimeFeedbackType.USER_TRANSCRIPTION.value,
-            self._user_speech_end_timestamp,
-            require_final=True,
-        )
-
-    def mark_bot_started_speaking(self, timestamp: Optional[str] = None):
-        """Record when the bot started speaking for the current assistant turn."""
-        self._bot_speech_start_timestamp = timestamp or self._now_iso()
-        self._bot_speech_end_timestamp = None
-        self._update_latest_payload_start_timestamp(
-            RealtimeFeedbackType.BOT_TEXT.value,
-            self._bot_speech_start_timestamp,
-        )
-
-    def mark_bot_stopped_speaking(self, timestamp: Optional[str] = None):
-        """Record when the bot stopped speaking and update the latest bot event."""
-        self._bot_speech_end_timestamp = timestamp or self._now_iso()
-        self._update_latest_payload_end_timestamp(
-            RealtimeFeedbackType.BOT_TEXT.value,
-            self._bot_speech_end_timestamp,
-        )
-
-    def _find_latest_open_payload(
-        self, event_type: str, *, require_final: bool = False
-    ) -> dict | None:
-        for event in reversed(self._events):
-            if event.get("type") != event_type:
-                continue
-            payload = event.get("payload")
-            if not isinstance(payload, dict):
-                continue
-            if require_final and payload.get("final") is not True:
-                continue
-            if payload.get("end_timestamp"):
-                continue
-            return payload
-        return None
-
-    def _update_latest_payload_start_timestamp(
-        self, event_type: str, start_timestamp: str, *, require_final: bool = False
-    ):
-        payload = self._find_latest_open_payload(
-            event_type, require_final=require_final
-        )
-        if payload is not None:
-            payload["timestamp"] = start_timestamp
-
-    def _update_latest_payload_end_timestamp(
-        self, event_type: str, end_timestamp: str, *, require_final: bool = False
-    ):
-        payload = self._find_latest_open_payload(
-            event_type, require_final=require_final
-        )
-        if payload is not None:
-            payload["end_timestamp"] = end_timestamp
-
-    def _event_with_speech_timestamps(self, event: dict) -> dict:
-        event_type = event.get("type")
-        payload = event.get("payload")
-        if not isinstance(payload, dict):
-            return event
-
-        payload_with_timestamps = dict(payload)
-        if (
-            event_type == RealtimeFeedbackType.USER_TRANSCRIPTION.value
-            and payload.get("final") is True
-        ):
-            if self._user_speech_start_timestamp:
-                payload_with_timestamps["timestamp"] = self._user_speech_start_timestamp
-            if self._user_speech_end_timestamp:
-                payload_with_timestamps["end_timestamp"] = (
-                    self._user_speech_end_timestamp
-                )
-        elif event_type == RealtimeFeedbackType.BOT_TEXT.value:
-            bot_interval_is_active = self._bot_speech_end_timestamp is None
-            if bot_interval_is_active and self._bot_speech_start_timestamp:
-                payload_with_timestamps["timestamp"] = self._bot_speech_start_timestamp
-
-        if payload_with_timestamps == payload:
-            return event
-        return {**event, "payload": payload_with_timestamps}
-
-    async def append(self, event: dict):
-        """Append a feedback event to the buffer with timestamp and current node."""
-        event = self._event_with_speech_timestamps(event)
+        """Append an immutable event with optional correlation metadata."""
+        if use_current_node:
+            node_id = self._current_node_id if node_id is None else node_id
+            node_name = self._current_node_name if node_name is None else node_name
         timestamped_event = stamp_realtime_feedback_event(
-            event,
-            timestamp=self._now_iso(),
-            turn=self._turn_counter,
-            node_id=self._current_node_id,
-            node_name=self._current_node_name,
+            deepcopy(event),
+            timestamp=timestamp or datetime.now(UTC).isoformat(timespec="milliseconds"),
+            turn=self._current_turn if turn is None else turn,
+            node_id=node_id,
+            node_name=node_name,
         )
         self._events.append(timestamped_event)
         logger.trace(
             f"Appended event {event.get('type')} to logs buffer for workflow {self._workflow_run_id}"
         )
 
-    def increment_turn(self):
-        """Increment turn counter (called on user transcription completion)."""
-        self._turn_counter += 1
-        logger.trace(
-            f"Incremented turn counter to {self._turn_counter} for workflow {self._workflow_run_id}"
-        )
-
     def _sorted_events(self) -> List[dict]:
-        # Stable sort by the realtime (payload) timestamp when available, falling
-        # back to the buffer-append timestamp. Python's sort is stable, so events
-        # sharing a key retain their original insertion order — this keeps
-        # consecutive bot-text chunks of a single turn contiguous.
+        # Stable sort by the top-level event timestamp used by the persisted
+        # realtime feedback schema. Legacy events without one fall back to their
+        # payload timestamp. Events sharing a key retain insertion order.
         return sorted(self._events, key=realtime_feedback_event_sort_key)
 
     def get_events(self) -> List[dict]:
-        """Get all events for final storage, ordered by realtime timestamp."""
+        """Get all events for final storage, ordered by event timestamp."""
         return self._sorted_events()
 
     def contains_user_speech(self) -> bool:

api/services/pipecat/realtime_feedback_events.py

@@ -166,4 +166,4 @@ def stamp_realtime_feedback_event(
 
 def realtime_feedback_event_sort_key(event: dict[str, Any]) -> str:
     payload_timestamp = (event.get("payload") or {}).get("timestamp")
-    return payload_timestamp or event.get("timestamp") or ""
+    return event.get("timestamp") or payload_timestamp or ""

api/services/pipecat/realtime_feedback_observer.py

@@ -21,7 +21,6 @@ node changes.
 """
 
 import json
-from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Awaitable, Callable, Optional, Set
 
 from loguru import logger
@@ -37,6 +36,9 @@ from api.services.pipecat.realtime_feedback_events import (
 
 if TYPE_CHECKING:
     from api.services.pipecat.in_memory_buffers import InMemoryLogsBuffer
+    from api.services.pipecat.transcript_log_coordinator import (
+        TranscriptLogCoordinator,
+    )
 
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
@@ -55,10 +57,6 @@ from pipecat.frames.frames import (
     TTSTextFrame,
     UserMuteStartedFrame,
     UserMuteStoppedFrame,
-    UserStartedSpeakingFrame,
-    UserStoppedSpeakingFrame,
-    VADUserStartedSpeakingFrame,
-    VADUserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import TTFBMetricsData
 from pipecat.observers.base_observer import BaseObserver, FramePushed
@@ -67,10 +65,6 @@ from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.utils.enums import RealtimeFeedbackType
 
 
-def _epoch_seconds_to_utc_iso(timestamp: float) -> str:
-    return datetime.fromtimestamp(timestamp, UTC).isoformat(timespec="milliseconds")
-
-
 class RealtimeFeedbackObserver(BaseObserver):
     """Observer that sends real-time events via WebSocket and persists final transcripts.
 
@@ -147,35 +141,13 @@ class RealtimeFeedbackObserver(BaseObserver):
             return
         # Bot speaking state - WS only (ephemeral state signals, not persisted)
         elif isinstance(frame, BotStartedSpeakingFrame):
-            if self._logs_buffer:
-                self._logs_buffer.mark_bot_started_speaking()
             await self._send_ws(
                 {"type": RealtimeFeedbackType.BOT_STARTED_SPEAKING.value, "payload": {}}
             )
         elif isinstance(frame, BotStoppedSpeakingFrame):
-            if self._logs_buffer:
-                self._logs_buffer.mark_bot_stopped_speaking()
             await self._send_ws(
                 {"type": RealtimeFeedbackType.BOT_STOPPED_SPEAKING.value, "payload": {}}
             )
-        elif isinstance(frame, UserStartedSpeakingFrame):
-            if self._logs_buffer:
-                self._logs_buffer.mark_user_started_speaking()
-        elif isinstance(frame, UserStoppedSpeakingFrame):
-            if self._logs_buffer:
-                self._logs_buffer.mark_user_stopped_speaking()
-        elif isinstance(frame, VADUserStartedSpeakingFrame):
-            if self._logs_buffer:
-                self._logs_buffer.mark_user_started_speaking(
-                    _epoch_seconds_to_utc_iso(frame.timestamp - frame.start_secs),
-                    from_vad=True,
-                )
-        elif isinstance(frame, VADUserStoppedSpeakingFrame):
-            if self._logs_buffer:
-                self._logs_buffer.mark_user_stopped_speaking(
-                    _epoch_seconds_to_utc_iso(frame.timestamp - frame.stop_secs),
-                    from_vad=True,
-                )
         # User mute state - WS only (ephemeral state signals, not persisted)
         elif isinstance(frame, UserMuteStartedFrame):
             await self._send_ws(
@@ -325,42 +297,36 @@ class RealtimeFeedbackObserver(BaseObserver):
 
 
 def register_turn_log_handlers(
-    logs_buffer: "InMemoryLogsBuffer",
+    transcript_coordinator: "TranscriptLogCoordinator",
     user_aggregator,
     assistant_aggregator,
 ):
     """Register event handlers on aggregators to persist final turn transcripts.
 
     Hooks into on_user_turn_message_added and on_assistant_turn_stopped to store
-    complete turn text in the logs buffer. Works for both WebRTC and telephony
-    calls — independent of WebSocket availability.
+    complete turn text through the turn-aware coordinator. Works for both
+    WebRTC and telephony calls — independent of WebSocket availability.
     """
 
     @user_aggregator.event_handler("on_user_turn_message_added")
     async def on_user_turn_message_added(aggregator, message):
-        logs_buffer.increment_turn()
         try:
-            await logs_buffer.append(
-                build_user_transcription_event(
-                    text=message.content,
-                    final=True,
-                    timestamp=message.timestamp,
-                    end_timestamp=getattr(message, "end_timestamp", None),
-                )
+            await transcript_coordinator.record_user_transcript(
+                text=message.content,
+                timestamp=message.timestamp,
+                end_timestamp=getattr(message, "end_timestamp", None),
             )
         except Exception as e:
-            logger.error(f"Failed to append user turn to logs buffer: {e}")
+            logger.error(f"Failed to coordinate user turn transcript: {e}")
 
     @assistant_aggregator.event_handler("on_assistant_turn_stopped")
     async def on_assistant_turn_stopped(aggregator, message):
         if message.content:
             try:
-                await logs_buffer.append(
-                    build_bot_text_event(
-                        text=message.content,
-                        timestamp=message.timestamp,
-                        end_timestamp=getattr(message, "end_timestamp", None),
-                    )
+                await transcript_coordinator.record_assistant_transcript(
+                    text=message.content,
+                    timestamp=message.timestamp,
+                    end_timestamp=getattr(message, "end_timestamp", None),
                 )
             except Exception as e:
-                logger.error(f"Failed to append assistant turn to logs buffer: {e}")
+                logger.error(f"Failed to coordinate assistant turn transcript: {e}")

api/services/pipecat/run_pipeline.py

@@ -66,6 +66,7 @@ from api.services.pipecat.service_factory import (
 from api.services.pipecat.tracing_config import (
     ensure_tracing,
 )
+from api.services.pipecat.transcript_log_coordinator import TranscriptLogCoordinator
 from api.services.pipecat.transport_setup import create_webrtc_transport
 from api.services.pipecat.worker_runner import run_pipeline_worker
 from api.services.pipecat.ws_sender_registry import get_ws_sender
@@ -1033,6 +1034,12 @@ async def _run_pipeline_impl(
 
     # Create pipeline task with audio configuration
     task = create_pipeline_task(pipeline, workflow_run_id, audio_config)
+    transcript_log_coordinator = TranscriptLogCoordinator(in_memory_logs_buffer)
+    if task.turn_tracking_observer is None:
+        raise RuntimeError("Transcript logging requires turn tracking to be enabled")
+    transcript_log_coordinator.attach_turn_tracking_observer(
+        task.turn_tracking_observer
+    )
 
     for runtime_session in integration_runtime_sessions:
         runtime_session.attach(task)
@@ -1087,7 +1094,9 @@ async def _run_pipeline_impl(
 
     # Register turn log handlers for all call types (WebRTC and telephony)
     register_turn_log_handlers(
-        in_memory_logs_buffer, user_context_aggregator, assistant_context_aggregator
+        transcript_log_coordinator,
+        user_context_aggregator,
+        assistant_context_aggregator,
     )
 
     # Register event handlers — resolve provider_id for PostHog tracking
@@ -1101,6 +1110,7 @@ async def _run_pipeline_impl(
         engine=engine,
         audio_buffer=audio_buffer,
         in_memory_logs_buffer=in_memory_logs_buffer,
+        transcript_log_coordinator=transcript_log_coordinator,
         pipeline_metrics_aggregator=pipeline_metrics_aggregator,
         audio_config=audio_config,
         pre_call_fetch_task=pre_call_fetch_task,

api/services/pipecat/transcript_log_coordinator.py

@@ -0,0 +1,298 @@
+"""Turn-aware coordination for immutable persisted transcript events.
+
+The transcript text, speech timing, and logical turn lifecycle are produced by
+different parts of the pipeline and can arrive in either order. This module is
+the single place where those facts are joined. It emits a transcript event only
+after the owning logical turn has ended (or during a final flush), and never
+mutates an event after it has been appended to the logs buffer.
+"""
+
+import asyncio
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from api.services.pipecat.realtime_feedback_events import (
+    build_bot_text_event,
+    build_user_transcription_event,
+)
+
+if TYPE_CHECKING:
+    from api.services.pipecat.in_memory_buffers import InMemoryLogsBuffer
+    from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
+
+
+def _now_iso() -> str:
+    return datetime.now(UTC).isoformat(timespec="milliseconds")
+
+
+@dataclass
+class _TranscriptSide:
+    text: str | None = None
+    transcript_timestamp: str | None = None
+    event_timestamp: str | None = None
+    speech_start_timestamp: str | None = None
+    speech_end_timestamp: str | None = None
+    speaking: bool = False
+    emitted: bool = False
+    node_id: str | None = None
+    node_name: str | None = None
+
+
+@dataclass
+class _TurnTranscriptState:
+    turn_id: int
+    ended: bool = False
+    interrupted: bool = False
+    user: _TranscriptSide = field(default_factory=_TranscriptSide)
+    assistant: _TranscriptSide = field(default_factory=_TranscriptSide)
+
+
+class TranscriptLogCoordinator:
+    """Join turn, transcript, and speech facts before appending log events."""
+
+    def __init__(self, logs_buffer: "InMemoryLogsBuffer"):
+        self._logs_buffer = logs_buffer
+        self._states: dict[int, _TurnTranscriptState] = {}
+        self._active_turn_id: int | None = None
+        self._lock = asyncio.Lock()
+
+    def attach_turn_tracking_observer(self, observer: "TurnTrackingObserver") -> None:
+        """Subscribe to the canonical turn owner's correlated lifecycle events."""
+
+        @observer.event_handler("on_turn_started")
+        async def on_turn_started(_observer, turn_number: int):
+            await self.record_turn_started(turn_number)
+
+        @observer.event_handler("on_turn_ended")
+        async def on_turn_ended(
+            _observer,
+            turn_number: int,
+            _duration: float,
+            was_interrupted: bool,
+        ):
+            await self.record_turn_ended(turn_number, interrupted=was_interrupted)
+
+        @observer.event_handler("on_user_speech_started_for_turn")
+        async def on_user_speech_started_for_turn(_observer, turn_number: int, _data):
+            callback_timestamp = _now_iso()
+            await self.record_user_started_speaking(turn_number, callback_timestamp)
+
+        @observer.event_handler("on_user_speech_stopped_for_turn")
+        async def on_user_speech_stopped_for_turn(_observer, turn_number: int, _data):
+            callback_timestamp = _now_iso()
+            await self.record_user_stopped_speaking(turn_number, callback_timestamp)
+
+        @observer.event_handler("on_bot_started_speaking")
+        async def on_bot_started_speaking(_observer, turn_number: int, _data):
+            await self.record_bot_started_speaking(turn_number)
+
+        @observer.event_handler("on_bot_stopped_speaking")
+        async def on_bot_stopped_speaking(_observer, turn_number: int, _data):
+            await self.record_bot_stopped_speaking(turn_number)
+
+    def _state(self, turn_id: int) -> _TurnTranscriptState:
+        state = self._states.get(turn_id)
+        if state is None:
+            state = _TurnTranscriptState(turn_id=turn_id)
+            self._states[turn_id] = state
+        return state
+
+    async def record_turn_started(self, turn_id: int) -> None:
+        async with self._lock:
+            self._state(turn_id)
+            if self._active_turn_id is None or turn_id >= self._active_turn_id:
+                self._active_turn_id = turn_id
+                self._logs_buffer.set_current_turn(turn_id)
+
+    async def record_turn_ended(self, turn_id: int, *, interrupted: bool) -> None:
+        async with self._lock:
+            state = self._state(turn_id)
+            state.ended = True
+            state.interrupted = interrupted
+            if self._active_turn_id == turn_id:
+                self._active_turn_id = None
+            await self._emit_ready_sides(state)
+
+    async def record_user_started_speaking(
+        self, turn_id: int, timestamp: str | None = None
+    ) -> None:
+        async with self._lock:
+            state = self._state(turn_id)
+            side = state.user
+            previous_start = side.speech_start_timestamp
+            candidate_start = timestamp or _now_iso()
+            side.speech_start_timestamp = (
+                min(previous_start, candidate_start)
+                if previous_start is not None
+                else candidate_start
+            )
+            side.speaking = True
+
+    async def record_user_stopped_speaking(
+        self, turn_id: int, timestamp: str | None = None
+    ) -> None:
+        async with self._lock:
+            side = self._state(turn_id).user
+            previous_end = side.speech_end_timestamp
+            candidate_end = timestamp or _now_iso()
+            side.speech_end_timestamp = (
+                max(previous_end, candidate_end)
+                if previous_end is not None
+                else candidate_end
+            )
+            side.speaking = False
+
+    async def record_bot_started_speaking(
+        self, turn_id: int, timestamp: str | None = None
+    ) -> None:
+        async with self._lock:
+            side = self._state(turn_id).assistant
+            if side.speech_start_timestamp is None:
+                side.speech_start_timestamp = timestamp or _now_iso()
+            side.speaking = True
+
+    async def record_bot_stopped_speaking(
+        self, turn_id: int, timestamp: str | None = None
+    ) -> None:
+        async with self._lock:
+            state = self._state(turn_id)
+            side = state.assistant
+            side.speech_end_timestamp = timestamp or _now_iso()
+            side.speaking = False
+            await self._emit_ready_sides(state)
+
+    async def record_user_transcript(
+        self,
+        *,
+        text: str,
+        timestamp: str | None,
+        end_timestamp: str | None = None,
+        event_timestamp: str | None = None,
+    ) -> None:
+        async with self._lock:
+            state = self._select_user_turn()
+            side = state.user
+            first_text = side.text is None
+            side.text = text if first_text else f"{side.text}\n{text}"
+            if first_text:
+                side.transcript_timestamp = timestamp
+                self._capture_node(side)
+            side.event_timestamp = event_timestamp or _now_iso()
+            if end_timestamp and not side.speech_end_timestamp:
+                side.speech_end_timestamp = end_timestamp
+            await self._emit_ready_sides(state)
+
+    async def record_assistant_transcript(
+        self,
+        *,
+        text: str,
+        timestamp: str | None,
+        end_timestamp: str | None = None,
+        event_timestamp: str | None = None,
+    ) -> None:
+        async with self._lock:
+            state = self._select_assistant_turn()
+            side = state.assistant
+            first_text = side.text is None
+            side.text = text if first_text else f"{side.text}\n{text}"
+            if first_text:
+                side.transcript_timestamp = timestamp
+                self._capture_node(side)
+            side.event_timestamp = event_timestamp or _now_iso()
+            if end_timestamp and not side.speech_end_timestamp:
+                side.speech_end_timestamp = end_timestamp
+            await self._emit_ready_sides(state)
+
+    def _select_user_turn(self) -> _TurnTranscriptState:
+        if self._active_turn_id is not None:
+            active = self._state(self._active_turn_id)
+            if active.user.text is None:
+                return active
+        candidates = [
+            state
+            for state in self._states.values()
+            if state.user.speech_start_timestamp and state.user.text is None
+        ]
+        if candidates:
+            return min(candidates, key=lambda state: state.turn_id)
+        if self._states:
+            return max(self._states.values(), key=lambda state: state.turn_id)
+        return self._state(1)
+
+    def _select_assistant_turn(self) -> _TurnTranscriptState:
+        candidates = [
+            state
+            for state in self._states.values()
+            if state.assistant.speech_start_timestamp and state.assistant.text is None
+        ]
+        interrupted = [
+            state for state in candidates if state.ended and state.interrupted
+        ]
+        if interrupted:
+            return min(interrupted, key=lambda state: state.turn_id)
+        if self._active_turn_id is not None:
+            active = self._state(self._active_turn_id)
+            if active.assistant.text is None:
+                return active
+        if candidates:
+            return min(candidates, key=lambda state: state.turn_id)
+        if self._states:
+            return max(self._states.values(), key=lambda state: state.turn_id)
+        return self._state(1)
+
+    def _capture_node(self, side: _TranscriptSide) -> None:
+        side.node_id = self._logs_buffer.current_node_id
+        side.node_name = self._logs_buffer.current_node_name
+
+    async def _emit_ready_sides(self, state: _TurnTranscriptState) -> None:
+        if not state.ended:
+            return
+        await self._emit_user(state)
+        if not state.assistant.speaking:
+            await self._emit_assistant(state)
+
+    async def _emit_user(self, state: _TurnTranscriptState) -> None:
+        side = state.user
+        if side.emitted or not side.text:
+            return
+        event = build_user_transcription_event(
+            text=side.text,
+            final=True,
+            timestamp=side.speech_start_timestamp or side.transcript_timestamp,
+            end_timestamp=side.speech_end_timestamp,
+        )
+        await self._append(state, side, event)
+
+    async def _emit_assistant(self, state: _TurnTranscriptState) -> None:
+        side = state.assistant
+        if side.emitted or not side.text:
+            return
+        event = build_bot_text_event(
+            text=side.text,
+            timestamp=side.speech_start_timestamp or side.transcript_timestamp,
+            end_timestamp=side.speech_end_timestamp,
+        )
+        await self._append(state, side, event)
+
+    async def _append(
+        self, state: _TurnTranscriptState, side: _TranscriptSide, event: dict
+    ) -> None:
+        await self._logs_buffer.append(
+            event,
+            timestamp=side.event_timestamp,
+            turn=state.turn_id,
+            node_id=side.node_id,
+            node_name=side.node_name,
+            use_current_node=False,
+        )
+        side.emitted = True
+
+    async def flush(self) -> None:
+        """Emit any remaining transcript text without inventing missing timing."""
+        async with self._lock:
+            for state in sorted(self._states.values(), key=lambda item: item.turn_id):
+                state.ended = True
+                state.user.speaking = False
+                state.assistant.speaking = False
+                await self._emit_ready_sides(state)