From 71899c249e445ce17f725a307946794bf50c4e12 Mon Sep 17 00:00:00 2001 From: Dumitru Gutu Date: Tue, 25 Aug 2026 11:08:12 +0100 Subject: [PATCH 1/6] RT SDK bugfixes and optimisations --- sdk/rt/speechmatics/rt/_async_client.py | 31 ++-- .../rt/_async_multi_channel_client.py | 30 ++-- sdk/rt/speechmatics/rt/_base_client.py | 92 +++++++++- sdk/rt/speechmatics/rt/_transport.py | 15 +- tests/rt/test_base_client.py | 164 ++++++++++++++++++ tests/rt/test_transport.py | 46 +++++ 6 files changed, 349 insertions(+), 29 deletions(-) create mode 100644 tests/rt/test_base_client.py create mode 100644 tests/rt/test_transport.py diff --git a/sdk/rt/speechmatics/rt/_async_client.py b/sdk/rt/speechmatics/rt/_async_client.py index e12e6a4e..79489224 100644 --- a/sdk/rt/speechmatics/rt/_async_client.py +++ b/sdk/rt/speechmatics/rt/_async_client.py @@ -10,7 +10,6 @@ from ._base_client import _BaseClient from ._exceptions import AudioError from ._exceptions import TimeoutError -from ._exceptions import TranscriptionError from ._logging import get_logger from ._models import AudioEncoding from ._models import AudioEventsConfig @@ -39,6 +38,9 @@ class AsyncClient(_BaseClient): url: WebSocket endpoint URL. If not provided, uses SPEECHMATICS_RT_URL environment variable or defaults to EU endpoint. conn_config: Websocket connection configuration. + sdk_identifier: Value reported to the service as `sm-sdk`. For a package built on + top of speechmatics-rt, pass its own identifier here instead of subclassing + Transport to override `_prepare_url`. Raises: ConfigurationError: If required configuration is missing or invalid. @@ -76,22 +78,26 @@ def __init__( api_key: Optional[str] = None, url: Optional[str] = None, conn_config: Optional[ConnectionConfig] = None, + sdk_identifier: Optional[str] = None, ) -> None: self._logger = get_logger("speechmatics.rt.async_client") + self._build_transport = lambda request_id: self._create_transport_from_config( + auth=auth, + api_key=api_key, + url=url, + conn_config=conn_config, + request_id=request_id, + sdk_identifier=sdk_identifier, + ) + ( self._session, self._recognition_started_evt, self._session_done_evt, ) = self._init_session_info() - transport = self._create_transport_from_config( - auth=auth, - api_key=api_key, - url=url, - conn_config=conn_config, - request_id=self._session.request_id, - ) + transport = self._build_transport(self._session.request_id) super().__init__(transport) self.on(ServerMessageType.RECOGNITION_STARTED, self._on_recognition_started) @@ -340,7 +346,12 @@ async def _send_eos(self, seq_no: int) -> None: async def _wait_recognition_started(self, timeout: float = 5.0) -> None: """Wait for RecognitionStarted message from server.""" - await asyncio.wait_for(self._recognition_started_evt.wait(), timeout) + await self._wait_started_or_session_done(self._recognition_started_evt, timeout) + + def _reset_session_events(self) -> None: + """Clear the session-tracking events, for a fresh session.""" + self._recognition_started_evt.clear() + self._session_done_evt.clear() def _on_recognition_started(self, msg: dict[str, Any]) -> None: """Handle RecognitionStarted message from server.""" @@ -357,8 +368,8 @@ def _on_error(self, msg: dict[str, Any]) -> None: """Handle Error message from server.""" error = msg.get("reason", "unknown") self._logger.error("Server error: %s", error) + self._last_error_reason = error self._session_done_evt.set() - raise TranscriptionError(error) def _on_audio_added(self, msg: dict[str, Any]) -> None: """Handle AudioAdded message from server.""" diff --git a/sdk/rt/speechmatics/rt/_async_multi_channel_client.py b/sdk/rt/speechmatics/rt/_async_multi_channel_client.py index 4be3b7b2..262255f9 100644 --- a/sdk/rt/speechmatics/rt/_async_multi_channel_client.py +++ b/sdk/rt/speechmatics/rt/_async_multi_channel_client.py @@ -10,7 +10,6 @@ from ._base_client import _BaseClient from ._exceptions import ConfigurationError from ._exceptions import TimeoutError -from ._exceptions import TranscriptionError from ._logging import get_logger from ._models import AudioEventsConfig from ._models import AudioFormat @@ -41,6 +40,8 @@ class AsyncMultiChannelClient(_BaseClient): url: WebSocket endpoint URL. If not provided, uses SPEECHMATICS_RT_URL environment variable or defaults to EU endpoint. conn_config: Websocket connection configuration. + sdk_identifier: Value reported to the service as `sm-sdk`, for a package built on + top of speechmatics-rt. Examples: Transcribing stereo audio: @@ -65,9 +66,19 @@ def __init__( api_key: Optional[str] = None, url: Optional[str] = None, conn_config: Optional[ConnectionConfig] = None, + sdk_identifier: Optional[str] = None, ) -> None: self._logger = get_logger("speechmatics.rt.async_multi_chan_client") + self._build_transport = lambda request_id: self._create_transport_from_config( + auth=auth, + api_key=api_key, + url=url, + conn_config=conn_config, + request_id=request_id, + sdk_identifier=sdk_identifier, + ) + ( self._session, self._rec_started_evt, @@ -75,13 +86,7 @@ def __init__( ) = self._init_session_info() self._eos_sent = False - transport = self._create_transport_from_config( - auth=auth, - api_key=api_key, - url=url, - conn_config=conn_config, - request_id=self._session.request_id, - ) + transport = self._build_transport(self._session.request_id) super().__init__(transport) @@ -182,7 +187,12 @@ async def transcribe( async def _wait_recognition_started(self, timeout: float = 5.0) -> None: """Wait for RecognitionStarted message from server.""" - await asyncio.wait_for(self._rec_started_evt.wait(), timeout=timeout) + await self._wait_started_or_session_done(self._rec_started_evt, timeout) + + def _reset_session_events(self) -> None: + """Clear the session-tracking events, for a fresh session.""" + self._rec_started_evt.clear() + self._session_done_evt.clear() async def _audio_producer(self, sources: dict[str, BinaryIO], chunk_size: int) -> None: """ @@ -247,8 +257,8 @@ def _on_eot(self, msg: dict[str, Any]) -> None: def _on_error(self, msg: dict[str, Any]) -> None: """Handle Error message from server.""" + self._last_error_reason = msg.get("reason", "unknown") self._session_done_evt.set() - raise TranscriptionError(msg.get("reason", "unknown")) def _on_warning(self, msg: dict[str, Any]) -> None: """Handle Warning message from server.""" diff --git a/sdk/rt/speechmatics/rt/_base_client.py b/sdk/rt/speechmatics/rt/_base_client.py index b9fc4fac..d3965425 100644 --- a/sdk/rt/speechmatics/rt/_base_client.py +++ b/sdk/rt/speechmatics/rt/_base_client.py @@ -7,6 +7,7 @@ import os import uuid from typing import Any +from typing import Callable from typing import Optional from typing_extensions import Self @@ -14,6 +15,8 @@ from ._auth import AuthBase from ._auth import StaticKeyAuth from ._events import EventEmitter +from ._exceptions import TimeoutError +from ._exceptions import TranscriptionError from ._exceptions import TransportError from ._logging import get_logger from ._models import AudioEventsConfig @@ -45,6 +48,11 @@ def __init__(self, transport: Transport) -> None: self._eos_sent = False self._audio_bytes_sent = 0 self._seq_no = 0 + self._last_error_reason: Optional[str] = None + + self._session: SessionInfo + self._session_done_evt: asyncio.Event + self._build_transport: Callable[[str], Transport] self._logger = get_logger("speechmatics.rt.base_client") @@ -77,6 +85,7 @@ def _create_transport_from_config( url: Optional[str] = None, conn_config: Optional[ConnectionConfig] = None, request_id: Optional[str] = None, + sdk_identifier: Optional[str] = None, ) -> Transport: """ Create a Transport instance from common configuration parameters. @@ -87,6 +96,8 @@ def _create_transport_from_config( url: WebSocket URL or None for default conn_config: Connection configuration or None for default request_id: Request ID for debugging/tracking + sdk_identifier: Value reported to the service as `sm-sdk`, or None for this + package's own identifier Returns: Configured Transport instance @@ -96,7 +107,7 @@ def _create_transport_from_config( conn_config = conn_config or ConnectionConfig() request_id = request_id or str(uuid.uuid4()) - return Transport(url, conn_config, auth, request_id) + return Transport(url, conn_config, auth, request_id, sdk_identifier=sdk_identifier) async def _ws_connect(self, ws_headers: Optional[dict] = None) -> None: await self._transport.connect(ws_headers) @@ -135,6 +146,16 @@ def audio_bytes_sent(self) -> int: """Number of audio bytes sent to the server.""" return self._audio_bytes_sent + @property + def request_id(self) -> str: + """Client-generated id for this session, used for tracing.""" + return self._session.request_id + + @property + def session_id(self) -> Optional[str]: + """Service-assigned session id, set once RecognitionStarted arrives.""" + return self._session.session_id + async def send_message(self, message: dict[str, Any]) -> None: """ Send a message through the WebSocket. @@ -183,6 +204,30 @@ async def _recv_loop(self) -> None: finally: self._closed_evt.set() + def _reset_session_events(self) -> None: + """Clear the subclass-owned session-tracking events, for a fresh session.""" + raise NotImplementedError() + + def _begin_new_session(self) -> None: + """ + Reset per-connection state for a fresh session on a client that already ran one. + + Without this, a second start_session()/transcribe() after close() would reuse a + Transport that can never reconnect (Transport.close() is a one-way latch), under + the request_id of a session that already ended. + """ + self._session.request_id = str(uuid.uuid4()) + self._session.session_id = None + self._reset_session_events() + self._transport = self._build_transport(self._session.request_id) + self._recv_task = None + self._closed_evt = asyncio.Event() + self._eos_sent = False + self._audio_bytes_sent = 0 + self._seq_no = 0 + self._last_error_reason = None + self._logger.debug("Starting new session (request_id=%s)", self._session.request_id) + async def _start_recognition_session( self, *, @@ -192,6 +237,9 @@ async def _start_recognition_session( audio_events_config: Optional[AudioEventsConfig] = None, ws_headers: Optional[dict] = None, ) -> tuple[TranscriptionConfig, AudioFormat]: + if self._closed_evt.is_set(): + self._begin_new_session() + transcription_config = transcription_config or TranscriptionConfig() audio_format = audio_format or AudioFormat() @@ -207,9 +255,15 @@ async def _start_recognition_session( audio_events_config=audio_events_config, ) - await self._ws_connect(ws_headers) - await self.send_message(start_recognition_message) - await self._wait_recognition_started() + try: + await self._ws_connect(ws_headers) + await self.send_message(start_recognition_message) + await self._wait_recognition_started() + except BaseException: + # BaseException, not Exception: a caller's own timeout cancels this via CancelledError. + with contextlib.suppress(Exception): + await self.close() + raise return transcription_config, audio_format @@ -217,6 +271,33 @@ async def _wait_recognition_started(self, timeout: float = 5.0) -> None: """Wait for RecognitionStarted message from server.""" raise NotImplementedError() + async def _wait_started_or_session_done(self, started_evt: asyncio.Event, timeout: float) -> None: + """ + Wait for `started_evt`, but stop as soon as the session ends first. + + Error and EndOfTranscript only set `_session_done_evt`, so without this a rejection + before RecognitionStarted would report a bare timeout instead of why. + + Raises: + TranscriptionError: The session ended (Error or EndOfTranscript) before + RecognitionStarted, with the service's reason when one was reported. + TimeoutError: Neither happened within `timeout`. + """ + started = asyncio.create_task(started_evt.wait()) + session_done = asyncio.create_task(self._session_done_evt.wait()) + try: + done, _ = await asyncio.wait({started, session_done}, timeout=timeout, return_when=asyncio.FIRST_COMPLETED) + finally: + for task in (started, session_done): + if not task.done(): + task.cancel() + + if not done: + raise TimeoutError("Timed out waiting for RecognitionStarted") + if started_evt.is_set(): + return + raise TranscriptionError(self._last_error_reason or "Session ended before RecognitionStarted") + async def close(self) -> None: """ Gracefully close the client connection and clean up resources. @@ -225,7 +306,8 @@ async def close(self) -> None: if self._recv_task and not self._recv_task.done(): self._recv_task.cancel() - with contextlib.suppress(Exception): + # CancelledError is a BaseException (3.8+), so suppress(Exception) alone misses it. + with contextlib.suppress(Exception, asyncio.CancelledError): await asyncio.wait_for(self._recv_task, timeout=2.0) await self._transport.close() diff --git a/sdk/rt/speechmatics/rt/_transport.py b/sdk/rt/speechmatics/rt/_transport.py index 351e7694..0ed16f81 100644 --- a/sdk/rt/speechmatics/rt/_transport.py +++ b/sdk/rt/speechmatics/rt/_transport.py @@ -59,6 +59,8 @@ def __init__( conn_config: ConnectionConfig, auth: AuthBase, request_id: Optional[str] = None, + *, + sdk_identifier: Optional[str] = None, ) -> None: """ Initialize the transport with connection configuration. @@ -70,11 +72,15 @@ def __init__( auth: Authentication object containing credentials. request_id: Optional unique identifier for request tracking. Generated automatically if not provided. + sdk_identifier: Value reported to the service as `sm-sdk`. Defaults to this + package's own identifier; a package built on top of speechmatics-rt can pass + its own here instead of subclassing to override `_prepare_url`. """ self._url = url self._auth = auth self._conn_config = conn_config self._request_id = request_id or str(uuid.uuid4()) + self._sdk_identifier = sdk_identifier self._websocket: Optional[Union[ClientConnection, WebSocketClientProtocol]] = None self._closed = False self._logger = get_logger("speechmatics.rt.transport") @@ -112,6 +118,7 @@ async def connect(self, ws_headers: Optional[dict] = None) -> None: if ws_headers is None: ws_headers = {} + ws_headers.setdefault("X-Request-Id", self._request_id) ws_headers.update(await self._auth.get_auth_headers()) try: @@ -243,16 +250,16 @@ def _prepare_url(self) -> str: """ Prepare the WebSocket URL with SDK version information. - This method adds the SDK version as a query parameter to the WebSocket + This method adds the SDK identifier as a query parameter to the WebSocket URL for server-side tracking and debugging purposes. Returns: - The complete WebSocket URL with SDK version parameter. + The complete WebSocket URL with the sm-sdk parameter. """ parsed = urlparse(self._url) - query_params = dict(parse_qsl(parsed.query)) - query_params["sm-sdk"] = f"python-rt-sdk-v{get_version()}" + query_params = dict(parse_qsl(parsed.query, keep_blank_values=True)) + query_params["sm-sdk"] = self._sdk_identifier or f"python-rt-sdk-v{get_version()}" updated_query = urlencode(query_params) return urlunparse(parsed._replace(query=updated_query)) diff --git a/tests/rt/test_base_client.py b/tests/rt/test_base_client.py new file mode 100644 index 00000000..0b433a56 --- /dev/null +++ b/tests/rt/test_base_client.py @@ -0,0 +1,164 @@ +import asyncio + +import pytest +import pytest_asyncio + +from speechmatics.rt import AsyncClient +from speechmatics.rt import AsyncMultiChannelClient +from speechmatics.rt import ServerMessageType +from speechmatics.rt import TimeoutError as RTTimeoutError +from speechmatics.rt import TranscriptionError + +API_KEY = "test-key" + + +class StubTransport: + """Captures what the client sends instead of opening a WebSocket.""" + + def __init__(self): + self.sent = [] + self.closed = False + + async def send_message(self, payload): + self.sent.append(payload) + + async def close(self): + self.closed = True + + +class SessionStubTransport: + """A stub transport that accepts RecognitionStarted on connect, for a full session lifecycle.""" + + def __init__(self, request_id): + self.request_id = request_id + self.sent = [] + self.closed = False + self._queue = asyncio.Queue() + + async def connect(self, headers=None): + self._queue.put_nowait({"message": "RecognitionStarted", "id": f"session-{self.request_id}"}) + + async def send_message(self, payload): + self.sent.append(payload) + + async def receive_message(self): + return await self._queue.get() + + async def close(self): + self.closed = True + + +@pytest_asyncio.fixture +async def client(monkeypatch): + # Must construct AsyncClient (and its internal asyncio.Event()s) inside the test's own + # running loop: on Python 3.9, Event() binds to whatever loop is current at construction, + # and a plain sync fixture runs outside the loop pytest-asyncio sets up for the test. + monkeypatch.delenv("SPEECHMATICS_RT_URL", raising=False) + c = AsyncClient(api_key=API_KEY) + c._transport = StubTransport() + return c + + +def error_message(reason="Not Authorized", error_type="not_authorised"): + return {"message": ServerMessageType.ERROR, "type": error_type, "reason": reason} + + +@pytest.mark.asyncio +async def test_request_id_and_session_id_properties(client): + assert client.request_id == client._session.request_id + assert client.session_id is None + + client.emit(ServerMessageType.RECOGNITION_STARTED, {"message": "RecognitionStarted", "id": "s1"}) + assert client.session_id == "s1" + + +@pytest.mark.asyncio +async def test_wait_recognition_started_returns_once_started(client): + client.emit(ServerMessageType.RECOGNITION_STARTED, {"message": "RecognitionStarted", "id": "s1"}) + await client._wait_recognition_started(timeout=1.0) # must not raise or hang + + +@pytest.mark.asyncio +async def test_error_before_recognition_started_raises_transcription_error(client): + client.emit(ServerMessageType.ERROR, error_message()) + + with pytest.raises(TranscriptionError, match="Not Authorized"): + await client._wait_recognition_started(timeout=5.0) + + +@pytest.mark.asyncio +async def test_end_of_transcript_before_recognition_started_raises_transcription_error(client): + client.emit(ServerMessageType.END_OF_TRANSCRIPT, {"message": "EndOfTranscript"}) + + with pytest.raises(TranscriptionError, match="Session ended before RecognitionStarted"): + await client._wait_recognition_started(timeout=5.0) + + +@pytest.mark.asyncio +async def test_no_response_raises_the_typed_timeout_error(client): + with pytest.raises(RTTimeoutError): + await client._wait_recognition_started(timeout=0.05) + + +@pytest.mark.asyncio +async def test_failed_handshake_closes_the_transport(client): + """A session that never gets RecognitionStarted must not leak the connection or the + receive task - otherwise a caller retrying without an explicit close() first would reuse + a half-open transport instead of getting the fresh one _begin_new_session builds.""" + client._transport = SessionStubTransport(client.request_id) + client._transport.connect = lambda headers=None: asyncio.sleep(3600) # never responds + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(client._start_recognition_session(), timeout=0.05) + + assert client._closed_evt.is_set() + assert client._transport.closed + + +@pytest.mark.asyncio +async def test_close_after_error_does_not_leak_cancelled_error(client): + """The recv loop is still winding down when close() cancels it - that must not surface.""" + client._recv_task = asyncio.get_event_loop().create_task(asyncio.sleep(10)) + await client.close() # must not raise CancelledError + assert client._transport.closed + + +@pytest.mark.asyncio +async def test_second_session_gets_a_fresh_request_id_and_transport(client): + """Regression test: a client reused for a second session used to keep the first + session's request_id, and its Transport - permanently closed by the first close() - + could never reconnect at all.""" + client._transport = SessionStubTransport(client.request_id) + client._build_transport = SessionStubTransport + + await client.start_session() + first_request_id = client.request_id + first_transport = client._transport + assert client.session_id == f"session-{first_request_id}" + + await client.close() + await client.start_session() + + assert client.request_id != first_request_id + assert client._transport is not first_transport + assert client.session_id == f"session-{client.request_id}" + + await client.close() + + +@pytest.mark.asyncio +async def test_multi_channel_client_second_session_gets_a_fresh_request_id(monkeypatch): + monkeypatch.delenv("SPEECHMATICS_RT_URL", raising=False) + client = AsyncMultiChannelClient(api_key=API_KEY) + client._transport = SessionStubTransport(client.request_id) + client._build_transport = SessionStubTransport + + await client._start_recognition_session() + first_request_id = client.request_id + + await client.close() + await client._start_recognition_session() + + assert client.request_id != first_request_id + + await client.close() diff --git a/tests/rt/test_transport.py b/tests/rt/test_transport.py new file mode 100644 index 00000000..d69c0a2b --- /dev/null +++ b/tests/rt/test_transport.py @@ -0,0 +1,46 @@ +import pytest + +import speechmatics.rt._transport as transport_module +from speechmatics.rt import ConnectionConfig +from speechmatics.rt import StaticKeyAuth +from speechmatics.rt._transport import Transport + + +class FakeWebSocket: + async def close(self): + pass + + +@pytest.mark.asyncio +async def test_request_id_is_sent_as_a_header(monkeypatch): + """The request_id we generate must actually reach the server, not just local logs.""" + captured = {} + + async def fake_connect(url, **kwargs): + captured["kwargs"] = kwargs + return FakeWebSocket() + + monkeypatch.setattr(transport_module, "connect", fake_connect) + + transport = Transport("wss://example.com/v2", ConnectionConfig(), StaticKeyAuth("key"), "my-request-id") + await transport.connect() + + headers = captured["kwargs"][transport_module.WS_HEADERS_KEY] + assert headers["X-Request-Id"] == "my-request-id" + + +@pytest.mark.asyncio +async def test_caller_supplied_request_id_header_is_not_overridden(monkeypatch): + captured = {} + + async def fake_connect(url, **kwargs): + captured["kwargs"] = kwargs + return FakeWebSocket() + + monkeypatch.setattr(transport_module, "connect", fake_connect) + + transport = Transport("wss://example.com/v2", ConnectionConfig(), StaticKeyAuth("key"), "my-request-id") + await transport.connect(ws_headers={"X-Request-Id": "caller-supplied"}) + + headers = captured["kwargs"][transport_module.WS_HEADERS_KEY] + assert headers["X-Request-Id"] == "caller-supplied" From bd6943d57be100bb86f582c1c74a5bb651986222 Mon Sep 17 00:00:00 2001 From: Dumitru Gutu Date: Tue, 25 Aug 2026 11:25:52 +0100 Subject: [PATCH 2/6] RT SDK bugfixes and optimisations --- .github/workflows/test.yaml | 10 +++++----- Makefile | 23 ++++++++++++++++++++++- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 7928371f..af37530b 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -22,7 +22,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: make install-dev + run: make install-dev-rt - name: Lint RT SDK run: make lint-rt - name: Test RT SDK @@ -42,7 +42,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: make install-dev + run: make install-dev-batch - name: Lint Batch SDK run: make lint-batch - name: Test Batch SDK @@ -62,7 +62,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: make install-dev + run: make install-dev-flow - name: Lint Flow SDK run: make lint-flow - name: Test Flow SDK @@ -82,7 +82,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: make install-dev + run: make install-dev-voice - name: Lint Voice Agent SDK run: make lint-voice - name: Test Voice Agent SDK @@ -102,7 +102,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: make install-dev + run: make install-dev-tts - name: Lint TTS SDK run: make lint-tts - name: Test TTS SDK diff --git a/Makefile b/Makefile index 12d4d8ae..da386b5d 100644 --- a/Makefile +++ b/Makefile @@ -7,6 +7,7 @@ .PHONY: type-check-all type-check-rt type-check-batch type-check-flow type-check-tts type-check-voice .PHONY: build-all build-rt build-batch build-flow build-tts build-voice .PHONY: clean-all clean-rt clean-batch clean-flow clean-tts clean-voice +.PHONY: install-dev install-dev-rt install-dev-batch install-dev-flow install-dev-tts install-dev-voice help: @@ -145,13 +146,33 @@ type-check-voice: cd sdk/voice/speechmatics && mypy . # Installation targets -install-dev: +# +# voice depends on speechmatics-rt>=0.5.3; the local rt package's own dev version (0.0.0) +# never satisfies that, so pip silently replaces the editable rt install with a published +# PyPI wheel the moment voice[dev] installs after it. rt must always be the LAST install +# in any sequence that also installs voice, so the local editable copy is what's left active. +install-dev: install-dev-batch install-dev-flow install-dev-tts install-dev-voice install-dev-rt + +install-dev-rt: python -m pip install --upgrade pip python -m pip install -e sdk/rt[dev] + +install-dev-batch: + python -m pip install --upgrade pip python -m pip install -e sdk/batch[dev] + +install-dev-flow: + python -m pip install --upgrade pip python -m pip install -e sdk/flow[dev] + +install-dev-tts: + python -m pip install --upgrade pip python -m pip install -e sdk/tts[dev] + +install-dev-voice: + python -m pip install --upgrade pip python -m pip install -e sdk/voice[dev] + python -m pip install -e sdk/rt install-build: python -m pip install --upgrade build From 1fc67b7bf0c215be684c530c1fb47fcace4cbc85 Mon Sep 17 00:00:00 2001 From: Dumitru Gutu Date: Tue, 25 Aug 2026 12:13:27 +0100 Subject: [PATCH 3/6] RT SDK bugfixes and optimisations --- .../rt/_async_multi_channel_client.py | 4 +- sdk/rt/speechmatics/rt/_base_client.py | 4 +- sdk/rt/speechmatics/rt/_events.py | 3 +- sdk/rt/speechmatics/rt/_transport.py | 3 +- tests/rt/test_base_client.py | 56 +++++++++++++++++++ tests/rt/test_transport.py | 26 +++++++++ 6 files changed, 91 insertions(+), 5 deletions(-) diff --git a/sdk/rt/speechmatics/rt/_async_multi_channel_client.py b/sdk/rt/speechmatics/rt/_async_multi_channel_client.py index 262255f9..753e0b36 100644 --- a/sdk/rt/speechmatics/rt/_async_multi_channel_client.py +++ b/sdk/rt/speechmatics/rt/_async_multi_channel_client.py @@ -257,7 +257,9 @@ def _on_eot(self, msg: dict[str, Any]) -> None: def _on_error(self, msg: dict[str, Any]) -> None: """Handle Error message from server.""" - self._last_error_reason = msg.get("reason", "unknown") + error = msg.get("reason", "unknown") + self._logger.error("Server error: %s", error) + self._last_error_reason = error self._session_done_evt.set() def _on_warning(self, msg: dict[str, Any]) -> None: diff --git a/sdk/rt/speechmatics/rt/_base_client.py b/sdk/rt/speechmatics/rt/_base_client.py index d3965425..8d96cbcf 100644 --- a/sdk/rt/speechmatics/rt/_base_client.py +++ b/sdk/rt/speechmatics/rt/_base_client.py @@ -54,7 +54,8 @@ def __init__(self, transport: Transport) -> None: self._session_done_evt: asyncio.Event self._build_transport: Callable[[str], Transport] - self._logger = get_logger("speechmatics.rt.base_client") + if not hasattr(self, "_logger"): + self._logger = get_logger("speechmatics.rt.base_client") @classmethod def _init_session_info(cls, request_id: Optional[str] = None) -> tuple[SessionInfo, asyncio.Event, asyncio.Event]: @@ -197,6 +198,7 @@ async def _recv_loop(self) -> None: except Exception as exc: self._logger.error("Receive loop error: %s", exc) self._closed_evt.set() + self._session_done_evt.set() try: await self._transport.close() except Exception: diff --git a/sdk/rt/speechmatics/rt/_events.py b/sdk/rt/speechmatics/rt/_events.py index 352f76dd..d5323541 100644 --- a/sdk/rt/speechmatics/rt/_events.py +++ b/sdk/rt/speechmatics/rt/_events.py @@ -20,7 +20,8 @@ class EventEmitter: def __init__(self) -> None: self._handlers: dict[ServerMessageType, set[Callable]] = {} self._once_handlers: dict[ServerMessageType, set[Callable]] = {} - self._logger = get_logger("speechmatics.rt.event_emitter") + if not hasattr(self, "_logger"): + self._logger = get_logger("speechmatics.rt.event_emitter") def on(self, event: ServerMessageType, callback: Optional[Callable] = None) -> Callable: """ diff --git a/sdk/rt/speechmatics/rt/_transport.py b/sdk/rt/speechmatics/rt/_transport.py index 0ed16f81..bd937bb0 100644 --- a/sdk/rt/speechmatics/rt/_transport.py +++ b/sdk/rt/speechmatics/rt/_transport.py @@ -116,8 +116,7 @@ async def connect(self, ws_headers: Optional[dict] = None) -> None: url_with_params = self._prepare_url() self._logger.debug("Connecting to WebSocket: %s", url_with_params) - if ws_headers is None: - ws_headers = {} + ws_headers = dict(ws_headers) if ws_headers else {} ws_headers.setdefault("X-Request-Id", self._request_id) ws_headers.update(await self._auth.get_auth_headers()) diff --git a/tests/rt/test_base_client.py b/tests/rt/test_base_client.py index 0b433a56..84c3aed9 100644 --- a/tests/rt/test_base_client.py +++ b/tests/rt/test_base_client.py @@ -8,6 +8,7 @@ from speechmatics.rt import ServerMessageType from speechmatics.rt import TimeoutError as RTTimeoutError from speechmatics.rt import TranscriptionError +from speechmatics.rt._exceptions import TransportError API_KEY = "test-key" @@ -26,6 +27,20 @@ async def close(self): self.closed = True +class RaisingTransport: + """A transport whose receive_message() fails immediately, simulating a dropped connection.""" + + def __init__(self, error=None): + self.error = error or TransportError("connection reset") + self.closed = False + + async def receive_message(self): + raise self.error + + async def close(self): + self.closed = True + + class SessionStubTransport: """A stub transport that accepts RecognitionStarted on connect, for a full session lifecycle.""" @@ -162,3 +177,44 @@ async def test_multi_channel_client_second_session_gets_a_fresh_request_id(monke assert client.request_id != first_request_id await client.close() + + +@pytest.mark.asyncio +async def test_async_client_logger_keeps_its_own_name(client): + """_BaseClient.__init__ must not clobber the subclass-specific logger AsyncClient set + before calling super().__init__() - otherwise per-logger filtering/level config aimed + at 'speechmatics.rt.async_client' silently gets no output.""" + assert client._logger.name == "speechmatics.rt.async_client" + + +@pytest.mark.asyncio +async def test_multi_channel_client_logger_keeps_its_own_name(monkeypatch): + monkeypatch.delenv("SPEECHMATICS_RT_URL", raising=False) + client = AsyncMultiChannelClient(api_key=API_KEY) + assert client._logger.name == "speechmatics.rt.async_multi_chan_client" + + +@pytest.mark.asyncio +async def test_recv_loop_error_unblocks_wait_for_recognition_started(client): + """A transport failure before RecognitionStarted must be reported as the real error + promptly, not swallowed until the full timeout elapses - _recv_loop's exception path + has to wake _wait_started_or_session_done via _session_done_evt, not just _closed_evt.""" + client._transport = RaisingTransport() + client._recv_task = asyncio.get_event_loop().create_task(client._recv_loop()) + + with pytest.raises(TranscriptionError): + await asyncio.wait_for(client._wait_recognition_started(timeout=5.0), timeout=1.0) + + assert client._closed_evt.is_set() + + +@pytest.mark.asyncio +async def test_multi_channel_on_error_logs_the_reason(caplog): + """AsyncMultiChannelClient._on_error must log server errors like AsyncClient._on_error + does, otherwise a failed multi-channel session leaves no trace before RecognitionStarted.""" + client = AsyncMultiChannelClient(api_key=API_KEY) + + with caplog.at_level("ERROR", logger="speechmatics.rt.async_multi_chan_client"): + client.emit(ServerMessageType.ERROR, error_message("Not Authorized")) + + assert "Not Authorized" in caplog.text diff --git a/tests/rt/test_transport.py b/tests/rt/test_transport.py index d69c0a2b..b7c1e5c6 100644 --- a/tests/rt/test_transport.py +++ b/tests/rt/test_transport.py @@ -44,3 +44,29 @@ async def fake_connect(url, **kwargs): headers = captured["kwargs"][transport_module.WS_HEADERS_KEY] assert headers["X-Request-Id"] == "caller-supplied" + + +@pytest.mark.asyncio +async def test_connect_does_not_mutate_caller_supplied_headers_dict(monkeypatch): + """A caller reusing the same ws_headers dict across sessions must get each session's + own X-Request-Id sent, not have the first session's id written into their dict and + silently reused (via setdefault no-op) on every later connect().""" + captured = [] + + async def fake_connect(url, **kwargs): + captured.append(dict(kwargs[transport_module.WS_HEADERS_KEY])) + return FakeWebSocket() + + monkeypatch.setattr(transport_module, "connect", fake_connect) + + shared_headers = {} + + transport1 = Transport("wss://example.com/v2", ConnectionConfig(), StaticKeyAuth("key"), "req-1") + await transport1.connect(ws_headers=shared_headers) + + transport2 = Transport("wss://example.com/v2", ConnectionConfig(), StaticKeyAuth("key"), "req-2") + await transport2.connect(ws_headers=shared_headers) + + assert shared_headers == {} + assert captured[0]["X-Request-Id"] == "req-1" + assert captured[1]["X-Request-Id"] == "req-2" From 5d7363e509fa4f44f8c77952a1f2a3ff00a7866f Mon Sep 17 00:00:00 2001 From: Dumitru Gutu Date: Tue, 25 Aug 2026 12:15:38 +0100 Subject: [PATCH 4/6] RT SDK bugfixes and optimisations --- sdk/voice/pyproject.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sdk/voice/pyproject.toml b/sdk/voice/pyproject.toml index 9006bd1f..4fdcf081 100644 --- a/sdk/voice/pyproject.toml +++ b/sdk/voice/pyproject.toml @@ -53,6 +53,8 @@ dev = [ "pytest-asyncio", "pytest-cov", "pytest-mock", + "aiofiles", + "types-aiofiles", "build", ] From 7d1959f2ce62a8ab71912acabc93e4322647615a Mon Sep 17 00:00:00 2001 From: Dumitru Gutu Date: Tue, 25 Aug 2026 13:35:22 +0100 Subject: [PATCH 5/6] RT SDK bugfixes and optimisations --- tests/rt/test_base_client.py | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/tests/rt/test_base_client.py b/tests/rt/test_base_client.py index 84c3aed9..8af384a8 100644 --- a/tests/rt/test_base_client.py +++ b/tests/rt/test_base_client.py @@ -78,15 +78,6 @@ def error_message(reason="Not Authorized", error_type="not_authorised"): return {"message": ServerMessageType.ERROR, "type": error_type, "reason": reason} -@pytest.mark.asyncio -async def test_request_id_and_session_id_properties(client): - assert client.request_id == client._session.request_id - assert client.session_id is None - - client.emit(ServerMessageType.RECOGNITION_STARTED, {"message": "RecognitionStarted", "id": "s1"}) - assert client.session_id == "s1" - - @pytest.mark.asyncio async def test_wait_recognition_started_returns_once_started(client): client.emit(ServerMessageType.RECOGNITION_STARTED, {"message": "RecognitionStarted", "id": "s1"}) @@ -187,13 +178,6 @@ async def test_async_client_logger_keeps_its_own_name(client): assert client._logger.name == "speechmatics.rt.async_client" -@pytest.mark.asyncio -async def test_multi_channel_client_logger_keeps_its_own_name(monkeypatch): - monkeypatch.delenv("SPEECHMATICS_RT_URL", raising=False) - client = AsyncMultiChannelClient(api_key=API_KEY) - assert client._logger.name == "speechmatics.rt.async_multi_chan_client" - - @pytest.mark.asyncio async def test_recv_loop_error_unblocks_wait_for_recognition_started(client): """A transport failure before RecognitionStarted must be reported as the real error From 3043b0cb3da3bc0ab1a8b15236cf233ca972f702 Mon Sep 17 00:00:00 2001 From: Dumitru Gutu Date: Tue, 25 Aug 2026 16:01:26 +0100 Subject: [PATCH 6/6] RT SDK bugfixes and optimisations --- sdk/rt/speechmatics/rt/_auth.py | 3 ++- sdk/rt/speechmatics/rt/_transport.py | 3 ++- sdk/rt/speechmatics/rt/constants.py | 1 + tests/rt/test_transport.py | 11 ++++++----- 4 files changed, 11 insertions(+), 7 deletions(-) diff --git a/sdk/rt/speechmatics/rt/_auth.py b/sdk/rt/speechmatics/rt/_auth.py index ee75bcac..ca0cc666 100644 --- a/sdk/rt/speechmatics/rt/_auth.py +++ b/sdk/rt/speechmatics/rt/_auth.py @@ -4,6 +4,7 @@ from typing import Optional from ._exceptions import AuthenticationError +from .constants import REQUEST_ID_HEADER class AuthBase(abc.ABC): @@ -125,7 +126,7 @@ async def _generate_token(self) -> str: } if self._request_id: - headers["X-Request-Id"] = self._request_id + headers.setdefault(REQUEST_ID_HEADER, self._request_id) try: async with aiohttp.ClientSession() as session: diff --git a/sdk/rt/speechmatics/rt/_transport.py b/sdk/rt/speechmatics/rt/_transport.py index bd937bb0..7b08de73 100644 --- a/sdk/rt/speechmatics/rt/_transport.py +++ b/sdk/rt/speechmatics/rt/_transport.py @@ -17,6 +17,7 @@ from ._logging import get_logger from ._models import ConnectionConfig from ._utils.version import get_version +from .constants import REQUEST_ID_HEADER try: # Try to import from new websockets >=13.0 @@ -117,7 +118,7 @@ async def connect(self, ws_headers: Optional[dict] = None) -> None: self._logger.debug("Connecting to WebSocket: %s", url_with_params) ws_headers = dict(ws_headers) if ws_headers else {} - ws_headers.setdefault("X-Request-Id", self._request_id) + ws_headers.setdefault(REQUEST_ID_HEADER, self._request_id) ws_headers.update(await self._auth.get_auth_headers()) try: diff --git a/sdk/rt/speechmatics/rt/constants.py b/sdk/rt/speechmatics/rt/constants.py index 649d6ce9..89d9aca5 100644 --- a/sdk/rt/speechmatics/rt/constants.py +++ b/sdk/rt/speechmatics/rt/constants.py @@ -1 +1,2 @@ CHUNK_SIZE = 4096 +REQUEST_ID_HEADER = "X-Request-Id" diff --git a/tests/rt/test_transport.py b/tests/rt/test_transport.py index b7c1e5c6..d0a8f7dc 100644 --- a/tests/rt/test_transport.py +++ b/tests/rt/test_transport.py @@ -4,6 +4,7 @@ from speechmatics.rt import ConnectionConfig from speechmatics.rt import StaticKeyAuth from speechmatics.rt._transport import Transport +from speechmatics.rt.constants import REQUEST_ID_HEADER class FakeWebSocket: @@ -26,7 +27,7 @@ async def fake_connect(url, **kwargs): await transport.connect() headers = captured["kwargs"][transport_module.WS_HEADERS_KEY] - assert headers["X-Request-Id"] == "my-request-id" + assert headers[REQUEST_ID_HEADER] == "my-request-id" @pytest.mark.asyncio @@ -40,10 +41,10 @@ async def fake_connect(url, **kwargs): monkeypatch.setattr(transport_module, "connect", fake_connect) transport = Transport("wss://example.com/v2", ConnectionConfig(), StaticKeyAuth("key"), "my-request-id") - await transport.connect(ws_headers={"X-Request-Id": "caller-supplied"}) + await transport.connect(ws_headers={REQUEST_ID_HEADER: "caller-supplied"}) headers = captured["kwargs"][transport_module.WS_HEADERS_KEY] - assert headers["X-Request-Id"] == "caller-supplied" + assert headers[REQUEST_ID_HEADER] == "caller-supplied" @pytest.mark.asyncio @@ -68,5 +69,5 @@ async def fake_connect(url, **kwargs): await transport2.connect(ws_headers=shared_headers) assert shared_headers == {} - assert captured[0]["X-Request-Id"] == "req-1" - assert captured[1]["X-Request-Id"] == "req-2" + assert captured[0][REQUEST_ID_HEADER] == "req-1" + assert captured[1][REQUEST_ID_HEADER] == "req-2"