feat(rt+voice): FEOU→EOU latency compensation

sdk/rt/speechmatics/rt/_async_client.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import math
 from typing import Any
 from typing import BinaryIO
 from typing import Optional
@@ -23,6 +24,43 @@
 
 _UNSET = object()
 
+# ForceEndOfUtterance latency-compensation grid.
+#
+# The engine decodes audio in fixed chunks and only resolves a forced EOU once
+# the chunk covering the FEOU timestamp is complete, so FEOU->EndOfUtterance
+# latency follows a sawtooth in the timestamp phase. Injecting silence up to the
+# next chunk boundary (plus a margin) before sending the FEOU gives the engine
+# the covering audio immediately and roughly halves the latency.
+#
+# Decode-chunk boundaries fall at FEOU_CHUNK_OFFSET + n * FEOU_CHUNK_PERIOD. The
+# values are empirical and may vary by engine version/endpoint; if the grid
+# differs only these constants change, not the algorithm. FEOU_MARGIN is the
+# head-room past the covering boundary (plateau 0.10-0.20s, 0.15 recommended).
+FEOU_CHUNK_PERIOD = 0.360
+FEOU_CHUNK_OFFSET = 0.290
+FEOU_MARGIN = 0.150
+
+# Server messages that carry audio-stream timestamps (start_time/end_time) somewhere
+# in their payload and so must be mapped back to real audio time when FEOU latency
+# compensation has injected silence (see AsyncClient._prepare_incoming_message).
+# High-frequency / timestamp-free messages (AudioAdded, RecognitionStarted, Info,
+# Warning, Error, EndOfTranscript, ...) are deliberately excluded.
+_TIMESTAMPED_SERVER_MESSAGES = frozenset(
+    {
+        ServerMessageType.ADD_TRANSCRIPT,
+        ServerMessageType.ADD_PARTIAL_TRANSCRIPT,
+        ServerMessageType.ADD_TRANSLATION,
+        ServerMessageType.ADD_PARTIAL_TRANSLATION,
+        ServerMessageType.END_OF_UTTERANCE,
+        ServerMessageType.AUDIO_EVENT_STARTED,
+        ServerMessageType.AUDIO_EVENT_ENDED,
+        ServerMessageType.SPEAKERS_RESULT,
+    }
+)
+
+# Payload keys that hold an audio-stream position in seconds.
+_TIMESTAMP_KEYS = ("start_time", "end_time")
+
 
 class AsyncClient(_BaseClient):
     """
@@ -102,6 +140,15 @@ def __init__(
 
         self._audio_format = AudioFormat(encoding=AudioEncoding.PCM_S16LE, sample_rate=44100, chunk_size=4096)
 
+        # ForceEndOfUtterance latency-compensation state.
+        # Silence injected ahead of real time to align a forced EOU with the engine's
+        # decode grid lengthens the server-side audio timeline, so returned timestamps
+        # run ahead of real audio time by the cumulative injected amount. We track the
+        # running total and an ordered list of (server_position, cumulative_injected)
+        # checkpoints so timestamps can be mapped back via adjust_timestamp.
+        self._injected_silence_seconds: float = 0.0
+        self._injected_silence_checkpoints: list[tuple[float, float]] = []
+
         self._logger.debug("AsyncClient initialized (request_id=%s)", self._session.request_id)
 
     async def start_session(
@@ -141,6 +188,11 @@ async def start_session(
         if audio_format is not None:
             self._audio_format = audio_format
 
+        # Reset per-session FEOU latency-compensation state so a reused client
+        # does not carry injected-silence bookkeeping across sessions.
+        self._injected_silence_seconds = 0.0
+        self._injected_silence_checkpoints = []
+
         await self._start_recognition_session(
             transcription_config=transcription_config,
             audio_format=audio_format,
@@ -169,7 +221,16 @@ async def stop_session(self) -> None:
         await self._session_done_evt.wait()  # Wait for end of transcript event to indicate we can stop listening
         await self.close()
 
-    async def force_end_of_utterance(self, *, timestamp: Optional[float] | object = _UNSET) -> None:
+    async def force_end_of_utterance(
+        self,
+        *,
+        timestamp: Optional[float] | object = _UNSET,
+        include_timestamp: bool = True,
+        compensate_latency: bool = False,
+        chunk_period: float = FEOU_CHUNK_PERIOD,
+        chunk_offset: float = FEOU_CHUNK_OFFSET,
+        margin: float = FEOU_MARGIN,
+    ) -> None:
         """
         This method sends a ForceEndOfUtterance message to the server to signal
         the end of an utterance. Forcing end of utterance will cause the final
@@ -179,8 +240,29 @@ async def force_end_of_utterance(self, *, timestamp: Optional[float] | object =
         to use for timing of the end of the utterance. If not provided, the timestamp
         will be calculated based on the cumulative audio sent to the server. If the provided
         timestamp is None, the ForceEndOfUtterance message will not include a timestamp.
+
+        When compensate_latency is True, silence is injected immediately (ahead of
+        real time) up to the next decode-chunk boundary past the marker plus a margin,
+        before the FEOU is sent. This gives the engine the audio covering the marker
+        straight away and roughly halves FEOU->EndOfUtterance latency. The FEOU
+        timestamp itself is left at the real speech-end marker (it is not moved to the
+        boundary). The injected silence lengthens the server-side audio timeline, so
+        subsequent returned timestamps must be mapped back with adjust_timestamp.
+
         Args:
-            timestamp: Optional timestamp for the request.
+            timestamp: Optional speech-end time in seconds, expressed in real audio time
+                (the audio you streamed, excluding any injected silence). It is shifted
+                onto the server timeline by the silence injected so far, so it lines up
+                with what the engine sees. If omitted, the current audio position is used.
+            include_timestamp: Whether to include the resolved timestamp in the message
+                (default True). When False, no timestamp is sent and the engine falls back
+                to its own timing; the marker is still computed for latency compensation.
+            compensate_latency: Inject silence to align the marker with the engine's
+                decode grid before sending the FEOU. Ignored when the timestamp is
+                omitted (explicit None).
+            chunk_period: Engine decode-chunk period in seconds (grid spacing).
+            chunk_offset: Decode-grid phase; boundaries fall at chunk_offset + n * chunk_period.
+            margin: Head-room in seconds injected past the covering chunk boundary.
 
         Raises:
             ConnectionError: If the WebSocket connection fails.
@@ -198,16 +280,164 @@ async def force_end_of_utterance(self, *, timestamp: Optional[float] | object =
 
         message: dict[str, Any] = {"message": ClientMessageType.FORCE_END_OF_UTTERANCE}
 
+        # Resolve the marker as a position on the server audio-stream timeline.
+        marker: Optional[float]
         if timestamp is _UNSET:
-            # default: auto-set from audio_seconds_sent
-            message["timestamp"] = self.audio_seconds_sent
+            # Default: the current audio position. audio_seconds_sent already includes any
+            # silence injected by earlier compensated FEOUs, so it is already server time.
+            marker = self.audio_seconds_sent
         elif timestamp is not None:
-            # user provided explicit value
-            message["timestamp"] = timestamp
-        # if timestamp is None: omit entirely
+            # Explicit value is a real-audio time; shift it onto the (possibly lengthened)
+            # server timeline by adding the silence injected so far. No-op until silence
+            # has been injected, so uncompensated usage is unaffected.
+            marker = float(timestamp) + self._injected_silence_seconds  # type: ignore[arg-type]
+        else:
+            # if timestamp is None: omit entirely
+            marker = None
+
+        # Inject the aligning silence before the FEOU. The marker is unchanged.
+        if compensate_latency and marker is not None:
+            await self._inject_feou_silence(
+                marker,
+                chunk_period=chunk_period,
+                chunk_offset=chunk_offset,
+                margin=margin,
+            )
+
+        if include_timestamp and marker is not None:
+            message["timestamp"] = marker
 
         await self.send_message(message)
 
+    async def _inject_feou_silence(
+        self,
+        marker: float,
+        *,
+        chunk_period: float,
+        chunk_offset: float,
+        margin: float,
+    ) -> None:
+        """
+        Inject silence up to the next decode-chunk boundary past a FEOU marker.
+
+        Sends a single block of PCM silence (all zero samples) through the same
+        outbound audio path as real audio, so the engine has the audio covering the
+        marker in hand and can resolve the forced EOU without waiting real-time for
+        the inter-utterance gap. The silence is sent in one message, not paced.
+
+        The injected silence extends the server-side audio timeline; the running total
+        and a (server_position, cumulative_injected) checkpoint are recorded so returned
+        timestamps can be mapped back to real audio time via adjust_timestamp.
+
+        Args:
+            marker: Speech-end position in seconds on the audio-stream timeline.
+            chunk_period: Engine decode-chunk period in seconds.
+            chunk_offset: Decode-grid phase; boundaries at chunk_offset + n * chunk_period.
+            margin: Head-room in seconds injected past the covering chunk boundary.
+
+        Raises:
+            ValueError: If the audio format is unset or has no encoding.
+        """
+        if self._audio_format is None:
+            raise ValueError("Cannot compensate FEOU latency before start_session sets an audio format")
+
+        sample_rate = self._audio_format.sample_rate
+        bytes_per_sample = self._audio_format.bytes_per_sample
+
+        # Next decode-chunk boundary at or after the marker (grid: offset + n * period).
+        boundary = chunk_offset + chunk_period * math.ceil((marker - chunk_offset) / chunk_period)
+        if boundary < marker:  # guard against float rounding leaving the boundary behind
+            boundary += chunk_period
+
+        pad_seconds = (boundary - marker) + margin
+        sample_count = round(pad_seconds * sample_rate)
+        if sample_count <= 0:
+            return
+
+        # int16/float32/mulaw zeros -> a run of zero bytes of the sample width.
+        await self._send_audio_bytes(bytes(sample_count * bytes_per_sample))
+
+        self._injected_silence_seconds += sample_count / sample_rate
+        self._injected_silence_checkpoints.append((self.audio_seconds_sent, self._injected_silence_seconds))
+
+    @property
+    def injected_silence_seconds(self) -> float:
+        """Cumulative silence injected for FEOU latency compensation, in seconds."""
+        return self._injected_silence_seconds
+
+    def adjust_timestamp(self, timestamp: float) -> float:
+        """
+        Map a server-timeline timestamp back to real audio time.
+
+        ForceEndOfUtterance latency compensation injects silence ahead of real time,
+        which lengthens the server-side audio timeline. As a result the start_time and
+        end_time on AddTranscript, AddPartialTranscript and EndOfUtterance messages run
+        ahead of real audio time by the amount of silence injected up to that position.
+        This subtracts that offset so the value lines up with the real audio the caller
+        streamed.
+
+        If no silence has been injected (compensation disabled or not yet triggered) the
+        timestamp is returned unchanged.
+
+        Args:
+            timestamp: A timestamp in seconds on the server audio timeline.
+
+        Returns:
+            The corresponding real audio time in seconds.
+        """
+        injected = 0.0
+        # Checkpoints are appended in increasing server_position order.
+        for server_position, cumulative in self._injected_silence_checkpoints:
+            if server_position <= timestamp:
+                injected = cumulative
+            else:
+                break
+        return timestamp - injected
+
+    def _prepare_incoming_message(self, msg: dict[str, Any]) -> None:
+        """
+        Map server-timeline timestamps on an incoming message back to real audio time.
+
+        When FEOU latency compensation has injected silence, the start_time/end_time
+        fields on transcript, translation, audio-event and end-of-utterance messages
+        run ahead of real audio time. This rewrites them in place (via adjust_timestamp)
+        before the message is emitted, so every listener sees real audio time without
+        needing to correct the values itself. It is a no-op until silence has been
+        injected.
+
+        Each timestamp is adjusted independently by its own audio-stream position, so a
+        single payload whose results span an injection point is handled correctly:
+        values before the injected silence keep the earlier offset, values after it get
+        the later (larger) offset.
+        """
+        # Nothing to correct until compensation has injected silence.
+        if not self._injected_silence_checkpoints:
+            return
+
+        if msg.get("message") not in _TIMESTAMPED_SERVER_MESSAGES:
+            return
+
+        self._adjust_timestamps_in_place(msg)
+
+    def _adjust_timestamps_in_place(self, node: Any) -> None:
+        """
+        Recursively map every audio-stream start_time/end_time within a payload back to
+        real audio time.
+
+        Walks nested dicts/lists so timestamps are corrected wherever they appear
+        (metadata, per-result timings, audio-event fields, ...) without hard-coding each
+        message's structure. Only numeric values are touched.
+        """
+        if isinstance(node, dict):
+            for key, value in node.items():
+                if key in _TIMESTAMP_KEYS and isinstance(value, (int, float)) and not isinstance(value, bool):
+                    node[key] = self.adjust_timestamp(value)
+                else:
+                    self._adjust_timestamps_in_place(value)
+        elif isinstance(node, list):
+            for item in node:
+                self._adjust_timestamps_in_place(item)
+
     @property
     def audio_seconds_sent(self) -> float:
         """Number of audio seconds sent to the server.

sdk/rt/speechmatics/rt/_base_client.py

@@ -116,12 +116,22 @@ async def send_audio(self, payload: bytes) -> None:
             >>> audio_chunk = b""
             >>> await client.send_audio(audio_chunk)
         """
-        if self._closed_evt.is_set() or self._eos_sent:
-            raise TransportError("Client is closed")
-
         if not isinstance(payload, bytes):
             raise ValueError("Payload must be bytes")
 
+        await self._send_audio_bytes(payload)
+
+    async def _send_audio_bytes(self, payload: bytes) -> None:
+        """
+        Write an audio payload to the transport and update the byte/sequence counters.
+
+        Shared by send_audio and internal audio injection (e.g. ForceEndOfUtterance
+        latency compensation) so both keep the server-side audio counters in sync
+        without passing through any subclass send_audio overrides.
+        """
+        if self._closed_evt.is_set() or self._eos_sent:
+            raise TransportError("Client is closed")
+
         try:
             await self._transport.send_message(payload)
             self._audio_bytes_sent += len(payload)
@@ -170,6 +180,7 @@ async def _recv_loop(self) -> None:
                 msg = await self._transport.receive_message()
 
                 if isinstance(msg, dict) and "message" in msg:
+                    self._prepare_incoming_message(msg)
                     self.emit(msg["message"], msg)
         except asyncio.CancelledError:
             pass
@@ -183,6 +194,16 @@ async def _recv_loop(self) -> None:
         finally:
             self._closed_evt.set()
 
+    def _prepare_incoming_message(self, msg: dict[str, Any]) -> None:
+        """
+        Hook to inspect or adjust an incoming server message before it is emitted.
+
+        Called for every server message with a "message" field, immediately before
+        it is dispatched to listeners. Subclasses may mutate the message in place
+        (for example to map timestamps back to real audio time). The base
+        implementation does nothing.
+        """
+
     async def _start_recognition_session(
         self,
         *,