chenghao-mou · Oct 13, 2026 · Oct 13, 2026 · Oct 13, 2026 · Oct 13, 2026 · Oct 13, 2026
@@ -359,8 +359,17 @@ async def _run(self) -> None:
             logger.error("all STTs are unavailable, retrying..")
 
         main_stream: RecognizeStream | None = None
+        # the leg that continues this stream's timeline; a leg created before the timeline
+        # has a start is seeded when its first frame is forwarded
+        seeded_stream: RecognizeStream | None = None
         forward_input_task: asyncio.Task[None] | None = None
 
+        def _seed(stream: RecognizeStream) -> None:
+            nonlocal seeded_stream
+            if (origin := self._input_ch.input_started_at) is not None:
+                stream.start_time_offset = time.time() - origin
+                seeded_stream = stream
+
         async def _forward_input_task() -> None:
             async for data in self._input_ch:
                 for stream in list(self._recovering_streams):
@@ -375,6 +384,8 @@ async def _forward_input_task() -> None:
                 if main_stream is not None:
                     try:
                         if isinstance(data, rtc.AudioFrame):
+                            if main_stream is not seeded_stream:
+                                _seed(main_stream)
                             main_stream.push_frame(data)
                         elif isinstance(data, self._FlushSentinel):
                             main_stream.flush()
@@ -400,10 +411,7 @@ async def _forward_input_task() -> None:
                             retry_interval=self._fallback_adapter._retry_interval,
                         ),
                     )
-                    # update main_stream start time offset so transcript timestamps are properly adjusted
-                    main_stream.start_time_offset = self.start_time_offset + (
-                        time.time() - self._start_time
-                    )
+                    _seed(main_stream)
 
                     if forward_input_task is None or forward_input_task.done():
                         forward_input_task = asyncio.create_task(_forward_input_task())

@@ -3,6 +3,7 @@
 import asyncio
 import time
 from abc import ABC, abstractmethod
+from collections import deque
 from collections.abc import AsyncIterable, AsyncIterator
 from dataclasses import dataclass, field
 from enum import Enum, unique
@@ -362,6 +363,69 @@ class _FlushSentinel:
 
         pass
 
+    class _InputChan(aio.Chan[rtc.AudioFrame | _FlushSentinel]):
+        """Input queue that places the provider's timestamps on the caller's timeline.
+
+        The provider counts audio from the first frame each connection reads, so the offset
+        is anchored on the capture time of that frame when it is read.
+        """
+
+        def __init__(self) -> None:
+            super().__init__()
+            # session start wall clock time; the first audio when unset
+            self.input_started_at: float | None = None
+            # session start -> capture of the current connection's first frame, or the value
+            # set before that frame is read (last write wins)
+            # word end time (connection relative) + offset = word end time (session relative)
+            self._start_time_offset = 0.0
+            # capture time of the first frame read since the provider's clock restarted
+            self._first_frame_at: float | None = None
+            # push time of each queued frame, in queue order
+            self._pushed_at: deque[float] = deque()
+            self._queued_duration = 0.0
+            self._last_pushed_at = 0.0
+
+        @property
+        def start_time_offset(self) -> float:
+            return self._start_time_offset
+
+        @start_time_offset.setter
+        def start_time_offset(self, value: float) -> None:
+            self.input_started_at = time.time() - value
+            self._start_time_offset = (
+                value
+                if self._first_frame_at is None
+                else self._first_frame_at - self.input_started_at
+            )
+
+        def restart_provider_clock(self) -> None:
+            """Anchor the offset again on the next frame read, for a new connection."""
+            self._first_frame_at = None
+
+        def send_nowait(self, value: rtc.AudioFrame | RecognizeStream._FlushSentinel) -> None:
+            super().send_nowait(value)
+            if isinstance(value, rtc.AudioFrame):
+                self._last_pushed_at = time.time()
+                self._pushed_at.append(self._last_pushed_at)
+                self._queued_duration += value.duration
+
+        def recv_nowait(self) -> rtc.AudioFrame | RecognizeStream._FlushSentinel:
+            value = super().recv_nowait()
+            if isinstance(value, rtc.AudioFrame):
+                pushed_at = self._pushed_at.popleft()
+                if self._first_frame_at is None:
+                    # a frame is pushed after it is captured, so its capture start is at most
+                    # its own push minus its duration (exact across pauses in the input) and
+                    # at most the last push minus all queued audio (exact for a burst)
+                    self._first_frame_at = min(
+                        pushed_at - value.duration, self._last_pushed_at - self._queued_duration
+                    )
+                    if self.input_started_at is None:
+                        self.input_started_at = self._first_frame_at
+                    self._start_time_offset = self._first_frame_at - self.input_started_at
+                self._queued_duration -= value.duration
+            return value
+
     def __init__(
         self,
         *,
@@ -379,7 +443,7 @@ def __init__(
         """
         self._stt = stt
         self._conn_options = conn_options
-        self._input_ch = aio.Chan[rtc.AudioFrame | RecognizeStream._FlushSentinel]()
+        self._input_ch = RecognizeStream._InputChan()
         self._event_ch = aio.Chan[SpeechEvent]()
 
         self._tee = aio.itertools.tee(self._event_ch, 2)
@@ -396,7 +460,6 @@ def __init__(
         self._pushed_sr = 0
         self._resampler: rtc.AudioResampler | None = None
 
-        self._start_time_offset: float = 0.0
         self._start_time: float = time.time()
         self._context: BaseModel | None = None
 
@@ -416,13 +479,20 @@ def context(self, value: BaseModel | None) -> None:
 
     @property
     def start_time_offset(self) -> float:
-        return self._start_time_offset
+        """Offset plugins add to the provider's timestamps.
+
+        Setting it says the caller's timeline began ``value`` seconds ago. With no value
+        set, the timeline begins with the stream's first audio. The provider counts from
+        the first frame it reads, and each new connection restarts that count, so the
+        offset is re-anchored on the first frame each attempt reads.
+        """
+        return self._input_ch.start_time_offset
 
     @start_time_offset.setter
     def start_time_offset(self, value: float) -> None:
         if value < 0:
             raise ValueError("start_time_offset must be non-negative")
-        self._start_time_offset = value
+        self._input_ch.start_time_offset = value
 
     @property
     def start_time(self) -> float:
@@ -464,17 +534,16 @@ async def _run(self) -> None: ...
 
     async def _main_task(self) -> None:
         max_retries = self._conn_options.max_retry
-        # we need to record last start time for each run/connection
-        # so that returned transcripts can have linear timestamps
         last_start_time = time.time()
 
         while self._num_retries <= max_retries:
             try:
-                self._start_time_offset += time.time() - last_start_time
                 self._start_time = time.time()
                 last_start_time = time.time()
                 return await self._run()
             except APIError as e:
+                # _run has stopped reading, so the next frame read is the next connection's
+                self._input_ch.restart_provider_clock()
                 # an attempt that outlived the connect timeout had connected, so this failure
                 # is not consecutive with the previous one and the budget starts over
                 if time.time() - last_start_time > self._conn_options.timeout:

@@ -387,6 +387,9 @@ def stt_node(
 
         You can override this node with your own implementation for more flexibility (e.g.,
         custom pre-processing of audio, additional buffering, or alternative STT strategies).
+        Timestamps in the yielded ``SpeechData`` are relative to the first audio frame the
+        node consumes, unless the node sets ``start_time_offset`` on its stream; the default
+        implementation sets it so they are relative to the session start.
 
         Args:
             audio (AsyncIterable[rtc.AudioFrame]): An asynchronous stream of audio frames.
@@ -525,10 +528,17 @@ async def stt_node(
             try:
                 conn_options = activity.session.conn_options.stt_conn_options
                 async with wrapped_stt.stream(conn_options=conn_options) as stream:
-                    _audio_input_started_at: float = (
-                        activity._audio_recognition._input_started_at
+                    # Timestamps are offset onto the session timeline: a new pipeline counts
+                    # from the recording or session start, a recreated stream continues the
+                    # pipeline's timeline. The pipeline anchors them on the same instant.
+                    pipeline = (
+                        activity._audio_recognition._stt_pipeline
                         if activity._audio_recognition is not None
-                        and activity._audio_recognition._input_started_at is not None
+                        else None
+                    )
+                    _audio_input_started_at: float = (
+                        pipeline.input_started_at
+                        if pipeline is not None and pipeline.input_started_at is not None
                         else (
                             activity.session._recorder_io.recording_started_at
                             if activity.session._recorder_io
@@ -539,6 +549,8 @@ async def stt_node(
                         )
                     )
                     stream.start_time_offset = time.time() - _audio_input_started_at
+                    if pipeline is not None:
+                        pipeline.input_started_at = _audio_input_started_at
 
                     @utils.log_exceptions(logger=logger)
                     async def _forward_input() -> None:

@@ -177,7 +177,8 @@ def __init__(
         self._event_ch = aio.Chan[stt.SpeechEvent]()
         self._pump_task = asyncio.create_task(self._stt_pump())
         self._pump_task.add_done_callback(lambda _: self._event_ch.close())
-        # wall-clock anchor for stream-based (STT and barge-in) timestamps
+        # copy of the session start wall clock time; the first audio when unset
+        # input_started_at + end_time (session relative) = wall-clock end of the speech
         self.input_started_at: float | None = None
 
     @property
@@ -767,7 +768,7 @@ def _push_audio(
         """
         self._sample_rate = frame.sample_rate
         if self._stt_pipeline is not None:
-            # stamp the wall-clock anchor on the first frame to reach the pipeline
+            # a node that declared no timeline counts from this frame (see input_started_at)
             if self._stt_pipeline.input_started_at is None:
                 self._stt_pipeline.input_started_at = time.time() - frame.duration
             self._stt_pipeline.audio_ch.send_nowait(stt_frame if stt_frame is not None else frame)

@@ -28,7 +28,7 @@ def _make_stream_under_test(
     instance._opts = SimpleNamespace(language="en-IN", sample_rate=16000)  # type: ignore[attr-defined]
     instance._speaking = False  # type: ignore[attr-defined]
     instance._should_flush = False  # type: ignore[attr-defined]
-    instance._start_time_offset = 0.0  # type: ignore[attr-defined]
+    instance._input_ch = SimpleNamespace(start_time_offset=0.0)  # type: ignore[attr-defined]
     instance._utterance_speech_start_wall = None  # type: ignore[attr-defined]
     instance._pending_final_data = None  # type: ignore[attr-defined]
     instance._pending_eos = False  # type: ignore[attr-defined]
@@ -120,7 +120,7 @@ async def test_final_timing_includes_start_time_offset() -> None:
     # Provider times are stream-relative; the base class accrues start_time_offset
     # across reconnects, so present speech_start/speech_end must be shifted by it.
     instance, captured = _make_stream_under_test()
-    instance._start_time_offset = 2.0  # type: ignore[attr-defined]
+    instance._input_ch = SimpleNamespace(start_time_offset=2.0)  # type: ignore[attr-defined]
 
     await instance._handle_events(_event("START_SPEECH"))
     await instance._handle_transcript_data(_ws_message(speech_start=0.5, speech_end=1.7))

@@ -124,7 +124,7 @@ def test_alignment_update_still_accounts_for_fallback(_fake_credentials: None) -
 def test_gateway_ink2_payload_carries_no_word_alignment() -> None:
     """The advertised word alignment is not in the data the gateway sends."""
     stream = InferenceSpeechStream.__new__(InferenceSpeechStream)
-    stream._start_time_offset = 0.0
+    stream._input_ch = SimpleNamespace(start_time_offset=0.0)
     stream._opts = SimpleNamespace(language="en")
 
     data = stream._build_speech_data(GATEWAY_INK2_FINAL)

@@ -28,7 +28,7 @@ def _stream() -> InferenceSpeechStream:
     stream._speaking = False
     stream._request_id = "req-1"
     stream._speech_duration = 0.0
-    stream._start_time_offset = 0.0
+    stream._input_ch = SimpleNamespace(start_time_offset=0.0)
     stream._pending_extra = None
     stream._opts = SimpleNamespace(language="en")
     return stream

@@ -21,7 +21,7 @@ def test_azure_recognized_emits_final_transcript(monkeypatch):
     stream._opts = SimpleNamespace(language=["en-US"])
     stream._loop = SimpleNamespace(call_soon_threadsafe=lambda callback, *args: callback(*args))
     stream._event_ch = SimpleNamespace(send_nowait=events.append)
-    stream._start_time_offset = 0.0
+    stream._input_ch = SimpleNamespace(start_time_offset=0.0)
 
     stream._on_recognized(
         SimpleNamespace(

@@ -67,7 +67,7 @@ def _new_stream(
     stream._event_ch = _EventSink()
     stream._speaking = False
     stream._last_partial_text = ""
-    stream._start_time_offset = 0.0
+    stream._input_ch = SimpleNamespace(start_time_offset=0.0)
     return stream