---

        await self._ensure_connected_for_call()
        self.dispatcher.register_subscription(scope, target, args)
        self.dispatcher.advance_watermark(scope)
        await self._send_keeping_connection(self._client.invoke, target, args)
        self.last_invocation_id = self._client.invocation_id

    async def invoke_stream(self, scope: SubscriptionScope, target: str, args: Any) -> None:
        """Отправляет streaming-вызов; в реестр не добавляется, потому что он одноразовый."""
        await self._ensure_connected_for_call()
        self.dispatcher.advance_watermark(scope)
        await self._send_keeping_connection(self._client.invoke_stream, target, args)
        self.last_invocation_id = self._client.invocation_id

    async def _send_keeping_connection(
        self, send_factory: Callable[..., Coroutine[Any, Any, Any]], target: str, args: Any
    ) -> None:
        """
        Отправляет вызов и переживает обрыв соединения между проверкой isConnected и send.
        """
        self._ensure_client()
        connection_before_send = self.connection_id
        try:
            await send_factory(target, args)
        except websockets.WebSocketException as error:
            logger.warning(
                f"[WS_MANAGER] Обрыв соединения при отправке {target}: {mask_secrets(error)}. "
                "Ждём, пока supervisor поднимет реконнект"
            )
            if not await self._wait_for_new_connection(connection_before_send, PersistConst.CALL_CONNECTION_WAIT_S):
                snapshot = self.health_snapshot()
                raise ConnectionError(
                    f"Соединение не восстановилось при отправке {target}: "
                    f"isConnected={snapshot['isConnected']}, connectionId={snapshot['connectionId']}, "
                    f"connectionsEstablished={snapshot['connectionsEstablished']}, "
                    f"connectFailures={snapshot['connectFailures']}"
                ) from error
            try:
                await send_factory(target, args)
            except websockets.WebSocketException as retry_error:
                snapshot = self.health_snapshot()
                raise ConnectionError(
                    f"Повторная отправка {target} не удалась: {mask_secrets(retry_error)}, "
                    f"isConnected={snapshot['isConnected']}, connectionId={snapshot['connectionId']}"
                ) from retry_error

    async def _wait_for_new_connection(self, previous_connection_id: Optional[str], timeout: float) -> bool:
        """
        Ждёт, пока supervisor поднимет соединение с новым connectionId.
        """
        deadline = time.monotonic() + timeout

        def connected_with_new_id() -> bool:
            return self._is_connected() and self.connection_id != previous_connection_id

        while time.monotonic() < deadline and not self._stopped.is_set():
            if connected_with_new_id():
                return True
            await asyncio.sleep(PersistConst.RECEIVE_POLL_INTERVAL_S)
        return connected_with_new_id()