diff --git a/sdk/rt/speechmatics/rt/_async_client.py b/sdk/rt/speechmatics/rt/_async_client.py index e12e6a4e..c1c30cab 100644 --- a/sdk/rt/speechmatics/rt/_async_client.py +++ b/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. diff --git a/sdk/rt/speechmatics/rt/_base_client.py b/sdk/rt/speechmatics/rt/_base_client.py index b9fc4fac..87b9ea58 100644 --- a/sdk/rt/speechmatics/rt/_base_client.py +++ b/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, *, diff --git a/sdk/voice/speechmatics/voice/_client.py b/sdk/voice/speechmatics/voice/_client.py index c0988dd3..474fdbdd 100644 --- a/sdk/voice/speechmatics/voice/_client.py +++ b/sdk/voice/speechmatics/voice/_client.py @@ -322,6 +322,11 @@ def __init__( self._forced_eou_active: bool = False self._last_forced_eou_latency: float = 0.0 + # Inject aligning silence before each forced EOU to cut its response latency. + # The RT client maps incoming server timestamps back to real audio time before + # they are emitted, so the rest of the pipeline is unaffected. + self._feou_latency_compensation: bool = self._config.feou_latency_compensation + # Emit EOT prediction (uses _uses_forced_eou) self._uses_eot_prediction: bool = self._eou_mode not in [ EndOfUtteranceMode.FIXED, @@ -1673,11 +1678,16 @@ async def _await_forced_eou(self, timeout: float = 1.0) -> None: # Received EOU eou_received: asyncio.Event = asyncio.Event() - # Add listener - self.once(AgentServerMessageType.END_OF_UTTERANCE, lambda message: eou_received.set()) + # Only the EndOfUtterance triggered by our ForceEndOfUtterance should satisfy + # the wait. The server marks these with `forced: true`, so a natural silence- + # based EOU that races in is ignored. A persistent listener is used (rather than + # `once`) so a non-forced EOU doesn't consume our one-shot handler; it is removed + # again on success or timeout. + def _on_end_of_utterance(message: dict[str, Any]) -> None: + if message.get("forced"): + eou_received.set() - # Trigger EOU message - self._emit_diagnostic_message("ForceEndOfUtterance sent - waiting for EndOfUtterance") + self.on(AgentServerMessageType.END_OF_UTTERANCE, _on_end_of_utterance) # Wait for EOU try: @@ -1686,7 +1696,10 @@ async def _await_forced_eou(self, timeout: float = 1.0) -> None: self._forced_eou_active = True # Send the force EOU and wait for the response - await self.force_end_of_utterance() + await self.force_end_of_utterance(compensate_latency=self._feou_latency_compensation) + self._emit_diagnostic_message("ForceEndOfUtterance sent - waiting for EndOfUtterance") + + # Wait for the response await asyncio.wait_for(eou_received.wait(), timeout=timeout) # Record the latency @@ -1696,6 +1709,7 @@ async def _await_forced_eou(self, timeout: float = 1.0) -> None: except asyncio.TimeoutError: pass finally: + self.off(AgentServerMessageType.END_OF_UTTERANCE, _on_end_of_utterance) self._forced_eou_active = False # ============================================================================ diff --git a/sdk/voice/speechmatics/voice/_models.py b/sdk/voice/speechmatics/voice/_models.py index b4a432c2..7feddd46 100644 --- a/sdk/voice/speechmatics/voice/_models.py +++ b/sdk/voice/speechmatics/voice/_models.py @@ -518,6 +518,12 @@ class VoiceAgentConfig(BaseModel): end of utterance detection and uses a fallback timer. Defaults to `EndOfUtteranceMode.FIXED`. + feou_latency_compensation: Whether to compensate for forced end of utterance latency. + When enabled, silence is injected ahead of real time to align each forced EOU with + the engine's decode grid, which roughly halves the forced EOU response time. + Returned timestamps are corrected back to real audio time automatically. + Defaults to `True`. + additional_vocab: List of additional vocabulary entries. If you supply a list of additional vocabulary entries, the this will increase the weight of the words in the vocabulary and help the STT engine to better transcribe the words. @@ -680,6 +686,7 @@ class VoiceAgentConfig(BaseModel): end_of_utterance_silence_trigger: float = 0.5 end_of_utterance_max_delay: float = 10.0 end_of_utterance_mode: EndOfUtteranceMode = EndOfUtteranceMode.FIXED + feou_latency_compensation: bool = True additional_vocab: list[AdditionalVocabEntry] = Field(default_factory=list) punctuation_overrides: Optional[dict] = None enable_entities: bool = False diff --git a/tests/voice/test_20_feou_latency.py b/tests/voice/test_20_feou_latency.py new file mode 100644 index 00000000..8772d18a --- /dev/null +++ b/tests/voice/test_20_feou_latency.py @@ -0,0 +1,257 @@ +import json +import os +import statistics +import time + +import pytest +from _utils import get_client +from _utils import send_audio_file +from pydantic import Field + +from speechmatics.rt import AsyncClient +from speechmatics.rt import AudioEncoding +from speechmatics.rt import AudioFormat +from speechmatics.voice import AgentServerMessageType +from speechmatics.voice._models import BaseModel +from speechmatics.voice._models import VoiceActivityConfig +from speechmatics.voice._models import VoiceAgentConfig +from speechmatics.voice._presets import VoiceAgentConfigPreset + +# Constants +API_KEY = os.getenv("SPEECHMATICS_API_KEY") +SHOW_LOG = os.getenv("SPEECHMATICS_SHOW_LOG", "0").lower() in ["1", "true"] + +# The live A/B latency test needs an API key and network; the timestamp unit test below +# is deterministic and offline, so gate only the live test rather than the whole module. +requires_live = pytest.mark.skipif( + API_KEY is None or os.getenv("CI") == "true", + reason="Requires a live API key and network; skipped in CI or when no key is set", +) + + +class TranscriptionSpeaker(BaseModel): + text: str + speaker_id: str = "S1" + start_time: float = 0.0 + end_time: float = 0.0 + + +class TranscriptionTest(BaseModel): + id: str + path: str + sample_rate: int + language: str + segments: list[TranscriptionSpeaker] = Field(default_factory=list) + + +# Audio file with several short utterances, giving one forced EOU per utterance +# and so multiple FEOU -> EndOfUtterance latency samples per run. +SAMPLE: TranscriptionTest = TranscriptionTest.from_dict( + { + "id": "08", + "path": "./assets/audio_08_16kHz.wav", + "sample_rate": 16000, + "language": "en", + "segments": [ + {"text": "Hello.", "start_time": 0.4, "end_time": 0.75}, + {"text": "Goodbye.", "start_time": 2.12, "end_time": 2.5}, + {"text": "Banana.", "start_time": 3.84, "end_time": 4.27}, + {"text": "Breakaway.", "start_time": 5.62, "end_time": 6.42}, + {"text": "Before.", "start_time": 7.76, "end_time": 8.16}, + {"text": "After.", "start_time": 9.56, "end_time": 10.05}, + ], + } +) + +# VAD silence duration (uses VAD end-of-turn detection, not smart turn). +VAD_DELAY_S = 0.18 + +# Endpoint +ENDPOINT = "wss://eu.rt.speechmatics.com/v2" + + +class LatencyRun(BaseModel): + """Result of a single transcription run.""" + + compensate: bool + latencies: list[float] = Field(default_factory=list) + segment_count: int = 0 + + @property + def mean_ms(self) -> float: + return statistics.mean(self.latencies) * 1000 if self.latencies else 0.0 + + +async def measure_run(endpoint: str, sample: TranscriptionTest, compensate: bool) -> LatencyRun: + """Run the sample once and measure per-utterance FEOU -> EndOfUtterance latency. + + Uses the ADAPTIVE preset (VAD-driven end of turn, no smart turn) and toggles + feou_latency_compensation so the same audio can be compared with and without + the fix. Latency is measured client-side as the wall-clock time between the + ForceEndOfUtterance being sent and the EndOfUtterance arriving. + """ + + # ADAPTIVE preset = VAD end-of-turn detection (no smart turn), with the + # compensation flag toggled for this run. + config = VoiceAgentConfigPreset.ADAPTIVE( + VoiceAgentConfig( + vad_config=VoiceActivityConfig(enabled=True, silence_duration=VAD_DELAY_S), + feou_latency_compensation=compensate, + ) + ) + + # Client + client = await get_client(url=endpoint, api_key=API_KEY, connect=False, config=config) + + # Run state + result = LatencyRun(compensate=compensate) + pending_sent: dict[str, float] = {} + + # Mark when a ForceEndOfUtterance is sent + def on_diagnostic(message): + if "ForceEndOfUtterance sent" in str(message.get("msg", "")): + pending_sent["t"] = time.perf_counter() + + # Pair the next EndOfUtterance with the last ForceEndOfUtterance sent + def on_end_of_utterance(message): + sent = pending_sent.pop("t", None) + if sent is not None: + result.latencies.append(time.perf_counter() - sent) + + # Count finalized segments (transcript integrity check) + def on_segment(message): + result.segment_count += len(message.get("segments", [])) + + # Listeners + client.on(AgentServerMessageType.DIAGNOSTICS, on_diagnostic) + client.on(AgentServerMessageType.END_OF_UTTERANCE, on_end_of_utterance) + client.on(AgentServerMessageType.ADD_SEGMENT, on_segment) + + # Connect + try: + await client.connect() + except Exception: + pytest.skip("Failed to connect to server") + + # Check we are connected + assert client._is_connected + + # Stream the audio at real time + await send_audio_file(client, sample.path) + + # Close session + await client.disconnect() + assert not client._is_connected + + return result + + +@pytest.mark.asyncio +@requires_live +async def test_feou_latency_compensation(): + """Compare FEOU -> EndOfUtterance latency with and without the compensation fix. + + Streams the same audio twice using VAD-driven end of turn: once with + feou_latency_compensation disabled (baseline) and once enabled (fix). The + injected silence should roughly halve the forced EOU response latency while + leaving the transcript unchanged. + """ + + # Baseline (no fix) then compensated (fix) + baseline = await measure_run(ENDPOINT, SAMPLE, compensate=False) + compensated = await measure_run(ENDPOINT, SAMPLE, compensate=True) + + # Summary (printed with pytest -s) + reduction = (1 - compensated.mean_ms / baseline.mean_ms) * 100 if baseline.mean_ms else 0.0 + print("\n=== FEOU -> EndOfUtterance latency ===") + print(f" baseline (no fix): {baseline.mean_ms:6.1f} ms over {len(baseline.latencies)} utterance(s)") + print(f" compensated (fix) : {compensated.mean_ms:6.1f} ms over {len(compensated.latencies)} utterance(s)") + print(f" reduction : {reduction:6.1f}%") + if SHOW_LOG: + print(f" baseline samples : {[round(x * 1000) for x in baseline.latencies]} ms") + print(f" compensated samples : {[round(x * 1000) for x in compensated.latencies]} ms") + print(f" segments (base/comp): {baseline.segment_count} / {compensated.segment_count}") + + # We need latency samples from both runs to compare + assert baseline.latencies, "No forced EOU latencies captured for the baseline run" + assert compensated.latencies, "No forced EOU latencies captured for the compensated run" + + # Transcript must be unaffected by the injected silence + assert baseline.segment_count > 0, "Baseline run produced no segments" + assert compensated.segment_count == baseline.segment_count, ( + f"Compensation changed the transcript: {compensated.segment_count} segments " + f"vs baseline {baseline.segment_count}" + ) + + # The compensation should reduce mean forced EOU latency + assert compensated.mean_ms < baseline.mean_ms, ( + f"Compensation did not reduce latency: {compensated.mean_ms:.1f} ms " + f"vs baseline {baseline.mean_ms:.1f} ms" + ) + + +class _FakeTransport: + """Captures outbound audio bytes and JSON messages without a network.""" + + def __init__(self): + self.messages: list[dict] = [] + self.audio_bytes: int = 0 + + async def send_message(self, data): + if isinstance(data, (bytes, bytearray)): + self.audio_bytes += len(data) + else: + self.messages.append(json.loads(data)) + + +def _offline_client(sample_rate: int = 16000) -> AsyncClient: + """An RT client wired to a fake transport, ready to send FEOUs offline.""" + client = AsyncClient(api_key="test") + client._transport = _FakeTransport() + client._audio_format = AudioFormat(encoding=AudioEncoding.PCM_S16LE, sample_rate=sample_rate) + return client + + +@pytest.mark.asyncio +async def test_explicit_feou_timestamp_shifted_onto_server_timeline(): + """A manually supplied FEOU timestamp is real-audio time and is mapped correctly. + + When compensation has injected silence, the server audio timeline runs ahead of the + real audio. An explicit timestamp (which the caller expresses in real audio time) must + be shifted onto the server timeline by the silence injected so far - the inverse of + adjust_timestamp - so it stays aligned across successive forced EOUs. This is offline + and deterministic (no network). + """ + rate, bps = 16000, 2 + client = _offline_client(rate) + + # 5.0s of real audio streamed; first compensated FEOU at real speech-end 5.0s. + # No silence has been injected yet, so the timestamp is sent unchanged. + client._audio_bytes_sent = int(5.0 * rate * bps) + await client.force_end_of_utterance(timestamp=5.0, compensate_latency=True) + assert client._transport.messages[-1]["timestamp"] == pytest.approx(5.0) + + injected = client.injected_silence_seconds + assert injected > 0, "compensation should have injected silence" + + # 3.0s more real audio; next FEOU at real speech-end 8.0s. The explicit real-audio + # time is shifted onto the server timeline by the silence injected so far. + client._audio_bytes_sent += int(3.0 * rate * bps) + await client.force_end_of_utterance(timestamp=8.0, compensate_latency=True) + sent = client._transport.messages[-1]["timestamp"] + assert sent == pytest.approx(8.0 + injected) + + # ...and it maps straight back to real audio time. + assert client.adjust_timestamp(sent) == pytest.approx(8.0) + + +@pytest.mark.asyncio +async def test_explicit_feou_timestamp_unchanged_without_injection(): + """Without injected silence an explicit timestamp is sent verbatim (backward compat).""" + rate, bps = 16000, 2 + client = _offline_client(rate) + client._audio_bytes_sent = int(4.0 * rate * bps) + + await client.force_end_of_utterance(timestamp=4.0, compensate_latency=False) + assert client._transport.messages[-1]["timestamp"] == pytest.approx(4.0) + assert client.injected_silence_seconds == 0.0