fix: redact smallest tts headers and report asr connection status

ai_agents/agents/integration_tests/asr_guarder/tests/test_connection_status.py

@@ -45,6 +45,7 @@
         "azure_asr_python",
         "bytedance_llm_based_asr",
         "deepgram_asr_python",
+        "smallest_asr_python",
         "soniox_asr_python",
         "tencent_asr_python",
     }

ai_agents/agents/ten_packages/extension/smallest_asr_python/extension.py

@@ -202,6 +202,11 @@ async def start_connection(self) -> None:
             self.audio_timeline.reset()
             self._utterance_start_ms = None
 
+            # Report the socket as usable now that the handshake succeeded,
+            # not just "connecting" — the base class only emits CONNECTING
+            # before this hook runs.
+            await self.on_connected()
+
             # Start message processing task
             self._message_task = asyncio.create_task(self._process_messages())
 
@@ -215,13 +220,13 @@ async def start_connection(self) -> None:
                 f"KEYPOINT start_connection failed: invalid vendor config: {e}"
             )
             self.connected = False
-            await self.send_asr_error(
-                ModuleError(
-                    module=MODULE_NAME_ASR,
-                    code=ModuleErrorCode.NON_FATAL_ERROR.value,
-                    message=str(e),
-                ),
+            error = ModuleError(
+                module=MODULE_NAME_ASR,
+                code=ModuleErrorCode.NON_FATAL_ERROR.value,
+                message=str(e),
             )
+            await self.send_asr_error(error)
+            await self.on_disconnected(code=error.code, message=error.message)
             self._schedule_reconnect()
 
     def _schedule_reconnect(self) -> None:
@@ -284,12 +289,14 @@ async def _process_messages(self) -> None:
                     )
                     # WebSocket closed unexpectedly, trigger reconnection
                     if not self.stopped:
-                        await self.send_asr_error(
-                            ModuleError(
-                                module=MODULE_NAME_ASR,
-                                code=ModuleErrorCode.NON_FATAL_ERROR.value,
-                                message=f"WebSocket closed unexpectedly: {msg.type}",
-                            ),
+                        error = ModuleError(
+                            module=MODULE_NAME_ASR,
+                            code=ModuleErrorCode.NON_FATAL_ERROR.value,
+                            message=f"WebSocket closed unexpectedly: {msg.type}",
+                        )
+                        await self.send_asr_error(error)
+                        await self.on_disconnected(
+                            code=error.code, message=error.message
                         )
                         # Schedule (do not await) so this task can exit before
                         # the reconnect path cancels it via stop_connection.
@@ -303,12 +310,14 @@ async def _process_messages(self) -> None:
             )
             if not self.stopped:
                 # Send error before attempting reconnection
-                await self.send_asr_error(
-                    ModuleError(
-                        module=MODULE_NAME_ASR,
-                        code=ModuleErrorCode.NON_FATAL_ERROR.value,
-                        message=f"WebSocket error, attempting reconnection: {str(e)}",
-                    ),
+                error = ModuleError(
+                    module=MODULE_NAME_ASR,
+                    code=ModuleErrorCode.NON_FATAL_ERROR.value,
+                    message=f"WebSocket error, attempting reconnection: {str(e)}",
+                )
+                await self.send_asr_error(error)
+                await self.on_disconnected(
+                    code=error.code, message=error.message
                 )
                 # Schedule (do not await): this runs inside `_message_task`,
                 # which the reconnect path cancels via stop_connection.

ai_agents/agents/ten_packages/extension/smallest_asr_python/tests/test_connection_status.py

@@ -0,0 +1,197 @@
+import json
+import threading
+from typing_extensions import override
+from ten_runtime import (
+    AsyncExtensionTester,
+    AsyncTenEnvTester,
+    Data,
+    TenError,
+    TenErrorCode,
+)
+
+# We must import it, which means this test fixture will be automatically executed
+from .mock import patch_smallest_ws  # noqa: F401
+
+
+class SmallestAsrConnectionStatusTester(AsyncExtensionTester):
+    """Collects every `connection_status_changed` transition observed."""
+
+    def __init__(self):
+        super().__init__()
+        self.transitions: list[dict] = []
+
+    def stop_test_if_checking_failed(
+        self,
+        ten_env_tester: AsyncTenEnvTester,
+        success: bool,
+        error_message: str,
+    ) -> None:
+        if not success:
+            err = TenError.create(
+                error_code=TenErrorCode.ErrorCodeGeneric,
+                error_message=error_message,
+            )
+            ten_env_tester.stop_test(err)
+
+    @override
+    async def on_data(
+        self, ten_env_tester: AsyncTenEnvTester, data: Data
+    ) -> None:
+        if data.get_name() != "connection_status_changed":
+            return
+        data_json, _ = data.get_property_to_json()
+        self.transitions.append(json.loads(data_json))
+
+
+# The handshake succeeds, so `on_connected()` must fire immediately after
+# `ws_connect()` returns — the base class only emits CONNECTING before
+# `start_connection()` runs, so without this the reported status would stay
+# stuck on "connecting" even though the socket is already usable.
+def test_connection_status_reports_connected_after_handshake(
+    patch_smallest_ws,
+):
+    def trigger_transcript_message():
+        transcript_message = {
+            "type": "transcription",
+            "transcript": "hello world",
+            "is_final": True,
+            "language": "en",
+        }
+        msg = patch_smallest_ws.MockWebSocketMessage(
+            msg_type=patch_smallest_ws.WSMsgType.TEXT,
+            data=json.dumps(transcript_message),
+        )
+        patch_smallest_ws.add_message(msg)
+
+    class ConnectedTester(SmallestAsrConnectionStatusTester):
+        @override
+        async def on_data(
+            self, ten_env_tester: AsyncTenEnvTester, data: Data
+        ) -> None:
+            await super().on_data(ten_env_tester, data)
+            if any(t.get("current") == "connected" for t in self.transitions):
+                ten_env_tester.stop_test()
+
+    def delayed_message_sender():
+        import time
+
+        time.sleep(0.5)
+        trigger_transcript_message()
+
+    threading.Thread(target=delayed_message_sender, daemon=True).start()
+
+    property_json = {
+        "params": {"api_key": "fake_api_key", "sample_rate": 16000}
+    }
+
+    tester = ConnectedTester()
+    tester.set_test_mode_single(
+        "smallest_asr_python", json.dumps(property_json)
+    )
+    err = tester.run()
+    assert err is None, (
+        f"test_connection_status_reports_connected_after_handshake err "
+        f"code: {err.error_code()} message: {err.error_message()}"
+    )
+    assert any(
+        t.get("current") == "connected" for t in tester.transitions
+    ), f"never observed a 'connected' transition: {tester.transitions}"
+
+
+# The vendor drops the socket mid-session (a CLOSED frame). The extension
+# must report "disconnected" (with close details) before scheduling the
+# reconnect — otherwise the reported connection_status stays wrong for the
+# whole reconnect window.
+def test_connection_status_reports_disconnected_on_ws_close(
+    patch_smallest_ws,
+):
+    connect_attempts = 0
+    transcript_message = {
+        "type": "transcription",
+        "transcript": "hello world",
+        "is_final": True,
+        "language": "en",
+    }
+
+    def push_after(delay, msg_type, data=None):
+        def _run():
+            import time
+
+            time.sleep(delay)
+            patch_smallest_ws.add_message(
+                patch_smallest_ws.MockWebSocketMessage(
+                    msg_type=msg_type, data=data
+                )
+            )
+
+        threading.Thread(target=_run, daemon=True).start()
+
+    from unittest.mock import patch
+
+    class MockSessionMidClose:
+        def __init__(self, *args, **kwargs) -> None:
+            self.closed: bool = False
+
+        async def ws_connect(self, url, headers=None, timeout=None):
+            nonlocal connect_attempts
+            connect_attempts += 1
+
+            ws = patch_smallest_ws.ws
+            ws.closed = False
+            ws._exception = None
+            with patch_smallest_ws.messages_lock:
+                patch_smallest_ws.messages.clear()
+
+            if connect_attempts == 1:
+                push_after(0.3, patch_smallest_ws.WSMsgType.CLOSED)
+            else:
+                push_after(
+                    0.3,
+                    patch_smallest_ws.WSMsgType.TEXT,
+                    json.dumps(transcript_message),
+                )
+            return ws
+
+        async def close(self) -> None:
+            self.closed = True
+
+    class DisconnectThenReconnectTester(SmallestAsrConnectionStatusTester):
+        @override
+        async def on_data(
+            self, ten_env_tester: AsyncTenEnvTester, data: Data
+        ) -> None:
+            await super().on_data(ten_env_tester, data)
+            if data.get_name() == "asr_result":
+                ten_env_tester.stop_test()
+
+    with patch(
+        "ten_packages.extension.smallest_asr_python.extension.aiohttp.ClientSession",
+        MockSessionMidClose,
+    ):
+        property_json = {
+            "params": {"api_key": "fake_api_key", "sample_rate": 16000}
+        }
+
+        tester = DisconnectThenReconnectTester()
+        tester.set_test_mode_single(
+            "smallest_asr_python", json.dumps(property_json)
+        )
+        err = tester.run()
+        assert err is None, (
+            f"test_connection_status_reports_disconnected_on_ws_close err "
+            f"code: {err.error_code()} message: {err.error_message()}"
+        )
+
+        statuses = [t.get("current") for t in tester.transitions]
+        assert (
+            "connected" in statuses
+        ), f"never observed a 'connected' transition: {tester.transitions}"
+        assert "disconnected" in statuses, (
+            "never observed a 'disconnected' transition after the ws close: "
+            f"{tester.transitions}"
+        )
+        # The close must be reported before the reconnect's own "connected"
+        # transition, not silently skipped.
+        assert statuses.index("disconnected") > statuses.index(
+            "connected"
+        ), f"disconnected did not follow the initial connect: {statuses}"

ai_agents/agents/ten_packages/extension/smallest_tts_python/config.py

@@ -85,6 +85,12 @@ def to_str(self, sensitive_handling: bool = True) -> str:
         if config.params and "api_key" in config.params:
             config.params["api_key"] = utils.encrypt(config.params["api_key"])
 
+        # Redact sensitive headers (e.g. Authorization) before logging —
+        # `headers.Authorization` is a supported auth path merged into the
+        # actual HTTP request, so it must not reach the key-point log in
+        # plaintext.
+        config.headers = utils.redact_headers(config.headers) or {}
+
         return f"{config}"
 
     def validate(self) -> None:

ai_agents/agents/ten_packages/extension/smallest_tts_python/tests/test_params.py

@@ -195,3 +195,56 @@ def test_validate():
     cfg.update_params()
     cfg.validate()  # should not raise
     print("✅ Validate test passed.")
+
+
+# ================ test to_str sensitive handling ================
+def test_to_str_masks_authorization_header():
+    """to_str() must not leak an Authorization header in plaintext.
+
+    headers.Authorization is a supported auth path merged into the actual
+    HTTP request, so the key-point log (which stringifies the whole config)
+    must not contain it verbatim.
+    """
+    from smallest_tts_python.config import SmallestTTSConfig
+
+    config = SmallestTTSConfig(
+        headers={"Authorization": "Bearer header_secret_token"},
+        params={"api_key": "test_api_key_123"},
+    )
+
+    rendered = config.to_str(sensitive_handling=True)
+
+    assert "Bearer header_secret_token" not in rendered
+    assert "test_api_key_123" not in rendered
+    print("Authorization header masking test passed.")
+
+
+def test_to_str_masks_api_key_header_variants():
+    """Case/variant coverage for the other header names Smallest accepts."""
+    from smallest_tts_python.config import SmallestTTSConfig
+
+    for header_name in ("api-key", "X-Api-Key", "xi-api-key"):
+        config = SmallestTTSConfig(headers={header_name: "secret_header_value"})
+
+        rendered = config.to_str(sensitive_handling=True)
+
+        assert (
+            "secret_header_value" not in rendered
+        ), f"{header_name} was not redacted"
+    print("API-key header variant masking test passed.")
+
+
+def test_to_str_without_sensitive_handling_keeps_raw_values():
+    """sensitive_handling=False is an explicit opt-out and must not redact."""
+    from smallest_tts_python.config import SmallestTTSConfig
+
+    config = SmallestTTSConfig(
+        headers={"Authorization": "Bearer header_secret_token"},
+        params={"api_key": "test_api_key_123"},
+    )
+
+    rendered = config.to_str(sensitive_handling=False)
+
+    assert "Bearer header_secret_token" in rendered
+    assert "test_api_key_123" in rendered
+    print("Non-sensitive to_str test passed.")