diff --git a/sdk/ai/azure-ai-agents/CHANGELOG.md b/sdk/ai/azure-ai-agents/CHANGELOG.md index 2d52918fa011..5dd9d2872c5a 100644 --- a/sdk/ai/azure-ai-agents/CHANGELOG.md +++ b/sdk/ai/azure-ai-agents/CHANGELOG.md @@ -10,7 +10,7 @@ - Added raw JSON WebSocket sends, complete unknown-event payloads, UTF-8 binary JSON reception, configurable receive limits and overflow policies, and opt-in recovery from malformed events. - Added custom WebSocket close codes and reasons, and per-event synchronous receive timeouts. -- Added realtime handshake options for session IDs, structured inputs, API versions, credential scopes, preview features, extra headers and query parameters, and same-host secure connection URL overrides. +- Added realtime session options for session IDs, structured inputs, persistence, and agent version selection. - Added preview `BetaVoiceAgentsTelephonyClient` and `BetaVoiceAgentsTelephonyAsyncClient` for outbound call jobs and campaign management, including recipient import, validation, publishing, pausing, resuming, and cancellation. - Added preview `BetaVoiceAgentsConversationsClient` and `BetaVoiceAgentsConversationsAsyncClient` for managing @@ -32,6 +32,8 @@ ### Bugs Fixed - Reject insecure voice-agent WebSocket URLs before token acquisition to prevent sending credentials over plaintext. +- Reject HTTP client, pipeline, policy, and retry builder settings that native WebSocket transports cannot honor. +- Enforce the single-iterator contract of synchronous voice-agent event streams. - Made synchronous voice-agent receive-buffer overflow signaling atomic across concurrent callbacks. - Fixed polling for telephony operations that return the `cancelled` status spelling. diff --git a/sdk/ai/azure-ai-agents/README.md b/sdk/ai/azure-ai-agents/README.md index 5cd2bd34ce0f..4593dfd20af0 100644 --- a/sdk/ai/azure-ai-agents/README.md +++ b/sdk/ai/azure-ai-agents/README.md @@ -120,22 +120,21 @@ ConversationService conversationService = openAIClient.conversations(); ### Realtime connection options Use `VoiceAgentWebSocketConnectionOptions` with the synchronous or asynchronous beta voice-agent client's -`openWebSocketSession` method to set session IDs, agent version overrides, structured inputs, API versions, credential -scopes, preview features, and extra handshake headers or query parameters. +`openWebSocketSession` method to configure session behavior such as session IDs, agent version selection, structured +inputs, persistence, buffering, and timeouts. ```java VoiceAgentWebSocketConnectionOptions options = new VoiceAgentWebSocketConnectionOptions() .setAgentSessionId("session-id") .setAgentVersionOverride("2") .setStructuredInputs("{\"language\":\"en\"}") - .setExtraHeaders(Collections.singletonMap("User-Agent", "my-application/1.0")); + .setStoreEnabled(true); ``` -Extra query parameters and non-protected headers override defaults. Authentication and WebSocket protocol -headers remain transport-controlled. An explicitly empty `Foundry-Features` value is preserved. -`setConnectionUrl` accepts a full `wss://` URI on the project endpoint's host and port, with no user information -or fragment. Existing query parameters are preserved unless overridden. URL validation happens before token -acquisition; cross-host overrides are rejected to prevent credentials from being sent to another host. +The SDK owns the WebSocket route, API version, authentication scope, transport, and preview feature headers. Endpoint, +credential, service version, configuration-based proxy settings, and `ClientOptions` are reused from +`AgentsClientBuilder`. Custom HTTP clients, pipelines, policies, and retry settings are rejected when building a +WebSocket client because the native WebSocket transports cannot apply them. ### Agent version drafts @@ -995,8 +994,8 @@ BetaVoiceAgentWebSocketAsyncClient realtimeAsyncClient #### Send a synchronous text turn -Connections require an `https://` or `wss://` project endpoint. Insecure endpoints and untrusted connection URL overrides -are rejected before acquiring a token. This also applies to localhost; use certificate-verified TLS for local servers. +Connections require an `https://` or `wss://` project endpoint. Insecure endpoints are rejected before acquiring a +token. This also applies to localhost; use certificate-verified TLS for local servers. Unknown server event types are returned as `RawRealtimeServerEvent`; `getRawEvent()` preserves the complete JSON object. Use `sendEvent(BinaryData)` to send raw JSON objects, including event types or fields not modeled by this SDK. Both diff --git a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/AgentsClientBuilder.java b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/AgentsClientBuilder.java index 8e7bede71955..aa2358145c62 100644 --- a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/AgentsClientBuilder.java +++ b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/AgentsClientBuilder.java @@ -678,8 +678,14 @@ public BetaMemoryStoresClient buildBetaMemoryStoresClient() { /** * Builds an asynchronous client for realtime voice-agent WebSocket sessions. + *

+ * Endpoint, credential, service version, configuration-based proxy settings, and client options are applied to + * WebSocket handshakes. Custom HTTP clients, pipelines, policies, and retry configuration are not compatible + * with the native WebSocket transport and cause this method to fail rather than being silently ignored. + * HTTP log options contribute to the user agent but do not configure WebSocket frame logging. * * @return an asynchronous voice-agent WebSocket client. + * @throws IllegalStateException if unsupported HTTP pipeline configuration is present. */ @Beta public BetaVoiceAgentWebSocketAsyncClient buildBetaVoiceAgentWebSocketAsyncClient() { @@ -688,8 +694,14 @@ public BetaVoiceAgentWebSocketAsyncClient buildBetaVoiceAgentWebSocketAsyncClien /** * Builds a synchronous client for realtime voice-agent WebSocket sessions. + *

+ * Endpoint, credential, service version, configuration-based proxy settings, and client options are applied to + * WebSocket handshakes. Custom HTTP clients, pipelines, policies, and retry configuration are not compatible + * with the native WebSocket transport and cause this method to fail rather than being silently ignored. + * HTTP log options contribute to the user agent but do not configure WebSocket frame logging. * * @return a synchronous voice-agent WebSocket client. + * @throws IllegalStateException if unsupported HTTP pipeline configuration is present. */ @Beta public BetaVoiceAgentWebSocketClient buildBetaVoiceAgentWebSocketClient() { @@ -821,10 +833,41 @@ private BetaVoiceAgentsTelephonyClient buildBetaVoiceAgentsTelephonyClient() { return new BetaVoiceAgentsTelephonyClient(buildInnerClient().getBetaVoiceAgentsTelephonies()); } + /** + * Creates the parallel configuration path required by the native WebSocket transports. Azure Core's + * {@link HttpClient} and {@link HttpPipeline} abstractions don't expose WebSocket session operations, so these + * clients can't reuse the generated HTTP pipeline directly. Compatible builder settings are adapted for the + * WebSocket handshake and must remain aligned with {@code createHttpPipeline()} when the TypeSpec emitter changes. + * HTTP transport, pipeline, policy, and retry settings are rejected rather than silently ignored. + * + * @return the voice-agent WebSocket client configuration. + * @throws IllegalStateException if unsupported HTTP pipeline configuration is present. + */ private VoiceAgentWebSocketClientConfiguration createVoiceAgentWebSocketConfiguration() { validateClient(); Objects.requireNonNull(tokenCredential, "'credential' must be configured to build a voice-agent WebSocket client."); + List unsupportedSettings = new ArrayList<>(); + if (httpClient != null) { + unsupportedSettings.add("httpClient"); + } + if (pipeline != null) { + unsupportedSettings.add("pipeline"); + } + if (!pipelinePolicies.isEmpty()) { + unsupportedSettings.add("addPolicy"); + } + if (retryOptions != null) { + unsupportedSettings.add("retryOptions"); + } + if (retryPolicy != null) { + unsupportedSettings.add("retryPolicy"); + } + if (!unsupportedSettings.isEmpty()) { + throw LOGGER.logExceptionAsError( + new IllegalStateException("Voice-agent WebSocket clients do not support these HTTP builder settings: " + + String.join(", ", unsupportedSettings) + ".")); + } Configuration buildConfiguration = configuration == null ? Configuration.getGlobalConfiguration() : configuration; ClientOptions localClientOptions = clientOptions == null ? new ClientOptions() : clientOptions; diff --git a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionAsyncClient.java b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionAsyncClient.java index 3dfe0a837b20..582be9c78089 100644 --- a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionAsyncClient.java +++ b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionAsyncClient.java @@ -118,7 +118,7 @@ Mono connect() { return Mono.error(new IllegalStateException("The voice-agent session has already been started.")); } - TokenRequestContext tokenContext = VoiceAgentWebSocketUtils.createTokenRequestContext(options); + TokenRequestContext tokenContext = VoiceAgentWebSocketUtils.createTokenRequestContext(); return configuration.getCredential().getToken(tokenContext).map(AccessToken::getToken).flatMap(token -> { Disposable connection = openWebSocket(token).subscribe(unused -> { }, this::terminateWithError, this::terminateNormally); diff --git a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClient.java b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClient.java index 525217e2249a..8310b946dafe 100644 --- a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClient.java +++ b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClient.java @@ -32,6 +32,8 @@ import com.azure.core.util.logging.ClientLogger; import java.io.IOException; import java.net.Proxy; +import java.net.ProxySelector; +import java.net.SocketAddress; import java.net.URI; import java.nio.charset.CharacterCodingException; import java.nio.charset.StandardCharsets; @@ -39,6 +41,7 @@ import java.util.Base64; import java.util.Collections; import java.util.Iterator; +import java.util.List; import java.util.NoSuchElementException; import java.util.Objects; import java.util.concurrent.ArrayBlockingQueue; @@ -48,6 +51,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; import okhttp3.Credentials; import okhttp3.OkHttpClient; import okhttp3.Request; @@ -87,7 +91,7 @@ private BetaVoiceAgentWebSocketSessionClient(VoiceAgentWebSocketClientConfigurat this.events = new ArrayBlockingQueue<>(receiveBufferCapacity + 1); this.websocketUri = VoiceAgentWebSocketUtils.buildWebSocketUri(configuration, agentName, options); String token = configuration.getCredential() - .getTokenSync(VoiceAgentWebSocketUtils.createTokenRequestContext(options)) + .getTokenSync(VoiceAgentWebSocketUtils.createTokenRequestContext()) .getToken(); this.httpClient = createHttpClient(configuration, options); Request.Builder request = new Request.Builder().url(websocketUri.toString()) @@ -173,7 +177,14 @@ public IterableStream receiveEvents(Duration timeout) { throw LOGGER.logExceptionAsError( new IllegalStateException("Only one receiveEvents iterator is supported per session.")); } - return IterableStream.of(() -> new EventIterator(events, timeout)); + AtomicBoolean iteratorCreated = new AtomicBoolean(); + return IterableStream.of(() -> { + if (!iteratorCreated.compareAndSet(false, true)) { + throw LOGGER.logExceptionAsError( + new IllegalStateException("The receiveEvents stream may only be iterated once.")); + } + return new EventIterator(events, timeout); + }); } /** @@ -445,7 +456,7 @@ private void shutdownHttpClient() { } } - private static OkHttpClient createHttpClient(VoiceAgentWebSocketClientConfiguration configuration, + static OkHttpClient createHttpClient(VoiceAgentWebSocketClientConfiguration configuration, VoiceAgentWebSocketConnectionOptions options) { OkHttpClient.Builder builder = new OkHttpClient.Builder(); builder.connectTimeout(options.getHandshakeTimeout().toMillis(), TimeUnit.MILLISECONDS) @@ -455,7 +466,29 @@ private static OkHttpClient createHttpClient(VoiceAgentWebSocketClientConfigurat if (proxyOptions != null) { Proxy.Type proxyType = proxyOptions.getType() == ProxyOptions.Type.SOCKS4 || proxyOptions.getType() == ProxyOptions.Type.SOCKS5 ? Proxy.Type.SOCKS : Proxy.Type.HTTP; - builder.proxy(new Proxy(proxyType, proxyOptions.getAddress())); + Proxy proxy = new Proxy(proxyType, proxyOptions.getAddress()); + if (proxyOptions.getNonProxyHosts() == null) { + builder.proxy(proxy); + } else { + Pattern nonProxyHosts = Pattern.compile(proxyOptions.getNonProxyHosts(), Pattern.CASE_INSENSITIVE); + builder.proxySelector(new ProxySelector() { + @Override + public List select(URI uri) { + return Collections + .singletonList(uri.getHost() != null && nonProxyHosts.matcher(uri.getHost()).matches() + ? Proxy.NO_PROXY + : proxy); + } + + @Override + public void connectFailed(URI uri, SocketAddress address, IOException error) { + LOGGER.atVerbose() + .addKeyValue("uri", uri) + .addKeyValue("proxyAddress", address) + .log("Failed to connect through the configured proxy."); + } + }); + } if (proxyOptions.getUsername() != null) { builder.proxyAuthenticator((route, response) -> response.request() .newBuilder() diff --git a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/implementation/realtime/VoiceAgentWebSocketUtils.java b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/implementation/realtime/VoiceAgentWebSocketUtils.java index ea921f92d37c..a1bb481c6ea7 100644 --- a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/implementation/realtime/VoiceAgentWebSocketUtils.java +++ b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/implementation/realtime/VoiceAgentWebSocketUtils.java @@ -3,9 +3,9 @@ package com.azure.ai.agents.implementation.realtime; +import com.azure.ai.agents.implementation.models.AgentDefinitionOptInKeys; import com.azure.ai.agents.models.RawRealtimeServerEvent; import com.azure.ai.agents.models.RealtimeServerEvent; -import com.azure.ai.agents.models.VoiceAgentTransport; import com.azure.ai.agents.models.VoiceAgentWebSocketConnectionOptions; import com.azure.core.credential.TokenRequestContext; import com.azure.core.http.HttpHeader; @@ -87,6 +87,7 @@ public static void validateClose(int code, String reason) { private static boolean isProtectedHeader(String name) { String lower = name.toLowerCase(Locale.ROOT); return "authorization".equals(lower) + || "user-agent".equals(lower) || "host".equals(lower) || "upgrade".equals(lower) || "connection".equals(lower) @@ -112,28 +113,10 @@ public static URI buildWebSocketUri(VoiceAgentWebSocketClientConfiguration confi String basePath = endpoint.getRawPath() == null ? "" : endpoint.getRawPath().replaceAll("/$", ""); String path = basePath + "/agents/" + encode(agentName) + "/endpoint/protocols/voice"; URI baseUri = URI.create(scheme + "://" + endpoint.getRawAuthority() + path); - if (options.getConnectionUrl() != null) { - baseUri = options.getConnectionUrl(); - int endpointPort = endpoint.getPort() == -1 ? 443 : endpoint.getPort(); - int overridePort = baseUri.getPort() == -1 ? 443 : baseUri.getPort(); - if (!"wss".equalsIgnoreCase(baseUri.getScheme()) - || baseUri.getHost() == null - || !baseUri.getHost().equalsIgnoreCase(endpoint.getHost()) - || endpointPort != overridePort - || baseUri.getRawUserInfo() != null - || baseUri.getRawFragment() != null) { - throw new IllegalArgumentException( - "Connection URL must be a wss URL on the project endpoint's host and port, without user information or a fragment."); - } - } UrlBuilder url = UrlBuilder.parse(baseUri.toString()); - url.setQueryParameter("api-version", - encode(options.getApiVersion() == null ? configuration.getApiVersion() : options.getApiVersion())); + url.setQueryParameter("api-version", encode(configuration.getApiVersion())); url.setQueryParameter("x-ms-client-sdk", encode(configuration.getUserAgent())); - VoiceAgentTransport transport = options.getTransport(); - if (transport != null) { - url.setQueryParameter("transport", encode(transport.toString())); - } + url.setQueryParameter("transport", "websocket"); if (options.isStoreEnabled() != null) { url.setQueryParameter("store", options.isStoreEnabled().toString()); } @@ -143,14 +126,11 @@ public static URI buildWebSocketUri(VoiceAgentWebSocketClientConfiguration confi if (options.getAgentSessionId() != null) { url.setQueryParameter("agent_session_id", encode(options.getAgentSessionId())); } - options.getExtraQuery().forEach((name, value) -> url.setQueryParameter(encode(name), encode(value))); return URI.create(url.toString()); } - public static TokenRequestContext createTokenRequestContext(VoiceAgentWebSocketConnectionOptions options) { - return options.getCredentialScopes() == null || options.getCredentialScopes().isEmpty() - ? new TokenRequestContext().addScopes(TOKEN_SCOPE) - : new TokenRequestContext().setScopes(options.getCredentialScopes()); + public static TokenRequestContext createTokenRequestContext() { + return new TokenRequestContext().addScopes(TOKEN_SCOPE); } public static HttpHeaders buildHeaders(VoiceAgentWebSocketClientConfiguration configuration, @@ -163,15 +143,11 @@ public static HttpHeaders buildHeaders(VoiceAgentWebSocketClientConfiguration co } } } - headers.set(HttpHeaderName.fromString("Foundry-Features"), options.getFoundryFeatures()); + headers.set(HttpHeaderName.fromString("Foundry-Features"), + AgentDefinitionOptInKeys.VOICE_AGENTS_V1_PREVIEW.toString()); if (options.getStructuredInputs() != null) { headers.set(HttpHeaderName.fromString("x-ms-voice-structured-inputs"), options.getStructuredInputs()); } - options.getExtraHeaders().forEach((name, value) -> { - if (!isProtectedHeader(name) || "Foundry-Features".equalsIgnoreCase(name)) { - headers.set(HttpHeaderName.fromString(name), value); - } - }); return headers.set(HttpHeaderName.AUTHORIZATION, "Bearer " + token); } diff --git a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/models/VoiceAgentWebSocketConnectionOptions.java b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/models/VoiceAgentWebSocketConnectionOptions.java index 23d29798891b..dba529e14b88 100644 --- a/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/models/VoiceAgentWebSocketConnectionOptions.java +++ b/sdk/ai/azure-ai-agents/src/main/java/com/azure/ai/agents/models/VoiceAgentWebSocketConnectionOptions.java @@ -6,13 +6,7 @@ import com.azure.ai.agents.implementation.utils.Beta; import com.azure.core.annotation.Fluent; import com.azure.core.util.logging.ClientLogger; -import java.net.URI; import java.time.Duration; -import java.util.ArrayList; -import java.util.Collections; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.function.Consumer; @@ -111,19 +105,12 @@ public VoiceAgentWebSocketConnectionOptions setMalformedEventHandler(Consumer credentialScopes; - private Map extraQuery = Collections.emptyMap(); - private Map extraHeaders = Collections.emptyMap(); /** * Creates options for opening a realtime voice-agent WebSocket session. @@ -143,21 +130,12 @@ public VoiceAgentWebSocketConnectionOptions(VoiceAgentWebSocketConnectionOptions this.maxMessageSize = source.maxMessageSize; this.overflowStrategy = source.overflowStrategy; this.malformedEventHandler = source.malformedEventHandler; - this.transport = source.transport; this.store = source.store; this.agentVersionOverride = source.agentVersionOverride; this.handshakeTimeout = source.handshakeTimeout; this.closeTimeout = source.closeTimeout; this.agentSessionId = source.agentSessionId; this.structuredInputs = source.structuredInputs; - this.connectionUrl = source.connectionUrl; - this.apiVersion = source.apiVersion; - this.foundryFeatures = source.foundryFeatures; - this.credentialScopes = source.credentialScopes == null - ? null - : Collections.unmodifiableList(new ArrayList<>(source.credentialScopes)); - this.extraQuery = Collections.unmodifiableMap(new LinkedHashMap<>(source.extraQuery)); - this.extraHeaders = Collections.unmodifiableMap(new LinkedHashMap<>(source.extraHeaders)); } /** @@ -196,142 +174,6 @@ public VoiceAgentWebSocketConnectionOptions setStructuredInputs(String structure return this; } - /** - * Gets the full WebSocket URL override. - * @return the URL override, or null. - */ - public URI getConnectionUrl() { - return connectionUrl; - } - - /** - * Sets a full WebSocket URL override. It must use wss and the project endpoint's host and port. - * User information and fragments are not supported. Existing query parameters are preserved unless overridden. - * @param connectionUrl the URL override, or null to use the agent route. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setConnectionUrl(URI connectionUrl) { - this.connectionUrl = connectionUrl; - return this; - } - - /** - * Gets the handshake API version override. - * @return the API version, or null. - */ - public String getApiVersion() { - return apiVersion; - } - - /** - * Sets the handshake API version override. - * @param apiVersion the API version, or null to use the client's version. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setApiVersion(String apiVersion) { - this.apiVersion = apiVersion; - return this; - } - - /** - * Gets the preview feature header value. - * @return the preview feature header value. - */ - public String getFoundryFeatures() { - return foundryFeatures; - } - - /** - * Sets the preview feature header value. - * @param foundryFeatures comma-separated preview features, or an empty string to suppress opt-in. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setFoundryFeatures(String foundryFeatures) { - this.foundryFeatures = Objects.requireNonNull(foundryFeatures, "'foundryFeatures' cannot be null."); - return this; - } - - /** - * Gets credential scopes for the handshake. - * @return an unmodifiable list, or null to use the default Foundry scope. - */ - public List getCredentialScopes() { - return credentialScopes; - } - - /** - * Sets credential scopes for the handshake. - * @param credentialScopes the scopes, or null to use the default Foundry scope. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setCredentialScopes(List credentialScopes) { - this.credentialScopes - = credentialScopes == null ? null : Collections.unmodifiableList(new ArrayList<>(credentialScopes)); - return this; - } - - /** - * Gets additional handshake query parameters. - * @return an unmodifiable map of query parameters. - */ - public Map getExtraQuery() { - return extraQuery; - } - - /** - * Sets additional handshake query parameters, taking precedence over defaults. - * @param extraQuery unencoded query names and values, or null to clear. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setExtraQuery(Map extraQuery) { - this.extraQuery = extraQuery == null - ? Collections.emptyMap() - : Collections.unmodifiableMap(new LinkedHashMap<>(extraQuery)); - return this; - } - - /** - * Gets additional handshake headers. - * @return an unmodifiable map of headers. - */ - public Map getExtraHeaders() { - return extraHeaders; - } - - /** - * Sets additional handshake headers. Authorization, host, connection, upgrade, and WebSocket protocol headers - * remain transport-controlled. Other headers override defaults case-insensitively, including empty values. - * @param extraHeaders the additional headers, or null to clear. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setExtraHeaders(Map extraHeaders) { - this.extraHeaders = extraHeaders == null - ? Collections.emptyMap() - : Collections.unmodifiableMap(new LinkedHashMap<>(extraHeaders)); - return this; - } - - /** - * Gets the session transport. - * - * @return the session transport. - */ - public VoiceAgentTransport getTransport() { - return transport; - } - - /** - * Sets the session transport. WebRTC transport performs signaling only; the SDK does not provide a WebRTC media - * implementation. - * - * @param transport the session transport. - * @return this options instance. - */ - public VoiceAgentWebSocketConnectionOptions setTransport(VoiceAgentTransport transport) { - this.transport = transport; - return this; - } - /** * Gets whether the conversation is persisted for this session. * diff --git a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClientTests.java b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClientTests.java new file mode 100644 index 000000000000..7cb270b780d9 --- /dev/null +++ b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/BetaVoiceAgentWebSocketSessionClientTests.java @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.ai.agents; + +import com.azure.ai.agents.implementation.realtime.VoiceAgentWebSocketClientConfiguration; +import com.azure.ai.agents.models.VoiceAgentWebSocketConnectionOptions; +import com.azure.core.http.ProxyOptions; +import java.net.InetSocketAddress; +import java.net.Proxy; +import java.net.URI; +import java.time.OffsetDateTime; +import java.util.List; +import okhttp3.OkHttpClient; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class BetaVoiceAgentWebSocketSessionClientTests { + @Test + public void syncTransportHonorsNonProxyHosts() { + InetSocketAddress proxyAddress = new InetSocketAddress("localhost", 8080); + ProxyOptions proxyOptions + = new ProxyOptions(ProxyOptions.Type.HTTP, proxyAddress).setNonProxyHosts("localhost"); + VoiceAgentWebSocketClientConfiguration configuration = new VoiceAgentWebSocketClientConfiguration( + URI.create("https://localhost"), + request -> Mono.just(new com.azure.core.credential.AccessToken("token", OffsetDateTime.now().plusHours(1))), + "v1", "test-user-agent", null, proxyOptions); + OkHttpClient client = BetaVoiceAgentWebSocketSessionClient.createHttpClient(configuration, + new VoiceAgentWebSocketConnectionOptions()); + + List bypassed = client.proxySelector().select(URI.create("https://LOCALHOST/session")); + List proxied = client.proxySelector().select(URI.create("https://example.com/session")); + + assertEquals(Proxy.NO_PROXY, bypassed.get(0)); + assertEquals(new Proxy(Proxy.Type.HTTP, proxyAddress), proxied.get(0)); + } +} diff --git a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/FoundryFeaturesHeaderVerificationTest.java b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/FoundryFeaturesHeaderVerificationTest.java index 93aa7b0bdf58..9d63a2e82cd7 100644 --- a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/FoundryFeaturesHeaderVerificationTest.java +++ b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/FoundryFeaturesHeaderVerificationTest.java @@ -14,7 +14,10 @@ import com.azure.core.http.HttpPipelineNextPolicy; import com.azure.core.http.HttpRequest; import com.azure.core.http.HttpResponse; +import com.azure.core.http.policy.ExponentialBackoffOptions; import com.azure.core.http.policy.HttpPipelinePolicy; +import com.azure.core.http.policy.RetryOptions; +import com.azure.core.http.policy.RetryPolicy; import com.azure.core.http.rest.RequestOptions; import com.azure.core.test.http.MockHttpResponse; import com.azure.core.test.utils.MockTokenCredential; @@ -362,6 +365,23 @@ private static String customPipelineHeader(RecordingHttpClient httpClient) { return httpClient.getLastRequest().getHeaders().getValue(CUSTOM_PIPELINE_HEADER); } + @Test + public void webSocketClientsRejectUnsupportedHttpConfiguration() { + RecordingHttpClient httpClient = new RecordingHttpClient(); + AgentsClientBuilder builder = createBuilder(createCustomPipeline(httpClient)).httpClient(httpClient) + .addPolicy(new CustomPipelinePolicy()) + .retryOptions(new RetryOptions(new ExponentialBackoffOptions())) + .retryPolicy(new RetryPolicy()); + + IllegalStateException syncError + = assertThrows(IllegalStateException.class, () -> builder.beta().buildBetaVoiceAgentWebSocketClient()); + assertTrue(syncError.getMessage().contains("httpClient, pipeline, addPolicy, retryOptions, retryPolicy")); + + IllegalStateException asyncError + = assertThrows(IllegalStateException.class, () -> builder.beta().buildBetaVoiceAgentWebSocketAsyncClient()); + assertEquals(syncError.getMessage(), asyncError.getMessage()); + } + private static HttpResponse openAIResponse(HttpRequest request) { String path = request.getUrl().getPath(); String responseBody = path.endsWith("/models") ? "{\"data\":[],\"object\":\"list\"}" : "{}"; diff --git a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentTelephonyLiveTests.java b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentTelephonyLiveTests.java index 56f57c25325f..f677e7174568 100644 --- a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentTelephonyLiveTests.java +++ b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentTelephonyLiveTests.java @@ -66,6 +66,10 @@ public class VoiceAgentTelephonyLiveTests { private static final Duration CALL_TIMEOUT = Duration.ofMinutes(2); private static final Duration POLL_INTERVAL = Duration.ofSeconds(2); + /** + * Validates the service lifecycle against an actual Twilio connection and phone number. The corresponding HTTP + * request and response contracts are covered without provider resources in {@link VoiceAgentTelephonyTests}. + */ @Test @EnabledIfEnvironmentVariable(named = "AZURE_TEST_MODE", matches = "LIVE") public void bindingLifecycleLive() { @@ -80,30 +84,38 @@ public void bindingLifecycleLive() { .allowPreview(true); AgentsClient agents = builder.buildAgentsClient(); BetaVoiceAgentsTelephonyClient telephony = builder.beta().buildBetaVoiceAgentsTelephonyClient(); - String agentName = "test-telephony-binding-" + shortId(); + String agentName = "tel-bind-" + shortId(); boolean agentCreated = false; + String bindingId = null; try { agents.createAgentVersion(agentName, new CreateAgentVersionInput(definition(model, "Greet the caller briefly, then say goodbye."))); agentCreated = true; TelephonyBinding binding = telephony.createTelephonyBinding(agentName, new CreateTwilioTelephonyBindingInput(connection, number).setLabel("Java SDK live test")); + bindingId = binding.getId(); TelephonyBindingListItem listedBinding = findBinding(telephony, agentName, binding.getId()); assertNotNull(listedBinding.getETag()); + String encodedBindingId = encodeBindingId(binding.getId()); - TelephonyBinding retrieved = telephony.getTelephonyBinding(agentName, binding.getId()); + TelephonyBinding retrieved = telephony.getTelephonyBinding(agentName, encodedBindingId); assertEquals(binding.getId(), retrieved.getId()); - TelephonyBinding updated = telephony.updateTelephonyBinding(agentName, binding.getId(), + TelephonyBinding updated = telephony.updateTelephonyBinding(agentName, encodedBindingId, listedBinding.getETag(), new UpdateTelephonyBindingInput().setLabel("Updated Java SDK live test")); assertEquals("Updated Java SDK live test", updated.getLabel()); String updatedEtag = findBinding(telephony, agentName, binding.getId()).getETag(); assertNotNull(updatedEtag); - telephony.deleteTelephonyBinding(agentName, binding.getId(), updatedEtag); + telephony.deleteTelephonyBinding(agentName, encodedBindingId, updatedEtag); + bindingId = null; assertTrue(telephony.listTelephonyBindings(agentName) .stream() .noneMatch(item -> binding.getId().equals(item.getId()))); } finally { + if (bindingId != null) { + String createdBindingId = bindingId; + safeCleanup("delete telephony binding", () -> deleteBinding(telephony, agentName, createdBindingId)); + } if (agentCreated) { safeCleanup("delete binding test agent", () -> agents.deleteAgent(agentName)); } @@ -127,12 +139,13 @@ public void twilioBindingAndOutboundCallLive() throws InterruptedException { .allowPreview(true); AgentsClient agents = builder.buildAgentsClient(); BetaVoiceAgentsTelephonyClient telephony = builder.beta().buildBetaVoiceAgentsTelephonyClient(); - String suffix = UUID.randomUUID().toString(); - String inboundAgent = "test-telephony-inbound-" + suffix; - String outboundAgent = "test-telephony-outbound-" + suffix; + String suffix = shortId(); + String inboundAgent = "tel-in-" + suffix; + String outboundAgent = "tel-out-" + suffix; String callJobId = null; String scheduledCallJobId = null; String inboundCallId = null; + String inboundBindingId = null; boolean inboundAgentCreated = false; boolean outboundAgentCreated = false; try { @@ -145,6 +158,7 @@ public void twilioBindingAndOutboundCallLive() throws InterruptedException { TelephonyBinding binding = telephony.createTelephonyBinding(inboundAgent, new CreateTwilioTelephonyBindingInput(connection1, number1).setLabel("Java SDK live test")); + inboundBindingId = binding.getId(); assertNotNull(binding.getId()); assertEquals(TelephonyProvider.TWILIO, binding.getProvider()); assertEquals(TelephonyBindingStatus.ACTIVE, binding.getStatus()); @@ -229,6 +243,11 @@ public void twilioBindingAndOutboundCallLive() throws InterruptedException { () -> telephony.replaceTelephonyTransferTargets(inboundAgent, getTransferTargetsEtag(telephony, inboundAgent), Collections.emptyList())); } + if (inboundBindingId != null) { + String bindingId = inboundBindingId; + safeCleanup("delete inbound telephony binding", + () -> deleteBinding(telephony, inboundAgent, bindingId)); + } if (outboundAgentCreated) { safeCleanup("delete outbound agent", () -> agents.deleteAgent(outboundAgent)); } @@ -294,6 +313,19 @@ private static TelephonyBindingListItem findBinding(BetaVoiceAgentsTelephonyClie .orElseThrow(() -> new AssertionError("Created binding was not listed.")); } + private static void deleteBinding(BetaVoiceAgentsTelephonyClient telephony, String agentName, String bindingId) { + telephony.listTelephonyBindings(agentName) + .stream() + .filter(item -> bindingId.equals(item.getId())) + .findFirst() + .ifPresent( + binding -> telephony.deleteTelephonyBinding(agentName, encodeBindingId(bindingId), binding.getETag())); + } + + private static String encodeBindingId(String bindingId) { + return bindingId.replace("+", "%2B"); + } + private static String getTransferTargetsEtag(BetaVoiceAgentsTelephonyClient telephony, String agentName) { return requireEtag(telephony.getTelephonyTransferTargetsWithResponse(agentName, new RequestOptions()), "telephony transfer targets"); diff --git a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionAsyncTests.java b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionAsyncTests.java new file mode 100644 index 000000000000..894b33800188 --- /dev/null +++ b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionAsyncTests.java @@ -0,0 +1,646 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.ai.agents.voice; + +import com.azure.ai.agents.AgentsClientBuilder; +import com.azure.ai.agents.BetaVoiceAgentWebSocketAsyncClient; +import com.azure.ai.agents.BetaVoiceAgentWebSocketSessionAsyncClient; +import com.azure.ai.agents.implementation.realtime.VoiceAgentWebSocketHttpResponse; +import com.azure.ai.agents.models.RawRealtimeServerEvent; +import com.azure.ai.agents.models.RealtimeResponseCreateEvent; +import com.azure.ai.agents.models.RealtimeSessionCreatedEvent; +import com.azure.ai.agents.models.RealtimeServerEvent; +import com.azure.ai.agents.models.VoiceAgentWarningEvent; +import com.azure.ai.agents.models.VoiceAgentWebSocketConnectionOptions; +import com.azure.ai.agents.models.VoiceAgentWebSocketOverflowStrategy; +import com.azure.core.credential.AccessToken; +import com.azure.core.credential.TokenCredential; +import com.azure.core.exception.ResourceModifiedException; +import com.azure.core.http.policy.HttpLogOptions; +import com.azure.core.util.BinaryData; +import com.azure.core.util.ClientOptions; +import com.azure.core.util.Configuration; +import com.azure.core.util.Header; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; +import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketFrame; +import java.net.URI; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.OffsetDateTime; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.ResourceLock; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.netty.DisposableServer; +import reactor.netty.http.server.WebsocketServerSpec; +import reactor.test.StepVerifier; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@ResourceLock("voice-agent-websocket-tls") +public class VoiceAgentWebSocketSessionAsyncTests { + private DisposableServer server; + + @Test + public void handshakeUsesSdkManagedProtocolValues() { + AtomicReference requestUri = new AtomicReference<>(); + AtomicReference headers = new AtomicReference<>(); + server = VoiceAgentWebSocketSessionTests.tlsServer().host("localhost").port(0).handle((request, response) -> { + requestUri.set(request.uri()); + headers.set(request.requestHeaders().copy()); + return response.sendWebsocket((inbound, outbound) -> inbound.receive().then(), + WebsocketServerSpec.builder().protocols("realtime").build()); + }).bindNow(); + AgentsClientBuilder builder + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project/") + .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))) + .configuration(Configuration.NONE) + .httpLogOptions(new HttpLogOptions().setApplicationId("log-app")) + .clientOptions(new ClientOptions().setApplicationId("client-app") + .setHeaders(Arrays.asList(new Header("X-Custom", "custom-value"), + new Header("Authorization", "Basic override"), new Header("User-Agent", "override-agent"), + new Header("Foundry-Features", "override-feature"), + new Header("Sec-WebSocket-Protocol", "override-protocol")))); + + BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() + .buildBetaVoiceAgentWebSocketAsyncClient() + .openWebSocketSession("agent name") + .block(Duration.ofSeconds(5)); + session.closeAsync().block(Duration.ofSeconds(5)); + assertFalse(session.isOpen()); + + assertEquals("Bearer test-token", headers.get().get(HttpHeaderNames.AUTHORIZATION)); + assertEquals("realtime", headers.get().get(HttpHeaderNames.SEC_WEBSOCKET_PROTOCOL)); + assertEquals("VoiceAgents=V1Preview", headers.get().get("Foundry-Features")); + assertEquals("custom-value", headers.get().get("X-Custom")); + assertEquals(1, headers.get().getAll(HttpHeaderNames.USER_AGENT).size()); + String userAgent = headers.get().get(HttpHeaderNames.USER_AGENT); + assertTrue(userAgent.startsWith("client-app azsdk-java-azure-ai-agents/"), userAgent); + assertFalse(userAgent.contains("log-app"), userAgent); + assertFalse(userAgent.contains("override-agent"), userAgent); + String uri = decode(requestUri.get()); + assertTrue(uri.contains("api-version=v1")); + assertTrue(uri.contains("transport=websocket")); + assertTrue(uri.contains("x-ms-client-sdk=" + userAgent)); + assertTrue(uri.startsWith("/api/projects/project/agents/agent name/endpoint/protocols/voice?")); + } + + @Test + public void handshakeUsesHttpLogApplicationIdFallback() { + AtomicReference requestUri = new AtomicReference<>(); + AtomicReference userAgent = new AtomicReference<>(); + server = VoiceAgentWebSocketSessionTests.tlsServer().host("localhost").port(0).handle((request, response) -> { + requestUri.set(request.uri()); + userAgent.set(request.requestHeaders().get(HttpHeaderNames.USER_AGENT)); + return response.sendWebsocket((inbound, outbound) -> inbound.receive().then(), + WebsocketServerSpec.builder().protocols("realtime").build()); + }).bindNow(); + AgentsClientBuilder builder + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") + .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))) + .configuration(Configuration.NONE) + .httpLogOptions(new HttpLogOptions().setApplicationId("log-app")) + .clientOptions(new ClientOptions()); + + BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() + .buildBetaVoiceAgentWebSocketAsyncClient() + .openWebSocketSession("agent") + .block(Duration.ofSeconds(5)); + session.closeAsync().block(Duration.ofSeconds(5)); + + assertTrue(userAgent.get().startsWith("log-app azsdk-java-azure-ai-agents/"), userAgent.get()); + assertTrue(decode(requestUri.get()).contains("x-ms-client-sdk=" + userAgent.get())); + } + + @BeforeAll + static void installTlsCertificate() { + VoiceAgentWebSocketSessionTests.installTlsCertificate(); + } + + @AfterEach + void disposeServer() { + if (server != null) { + server.disposeNow(); + } + } + + @AfterAll + static void restoreTlsConfiguration() { + VoiceAgentWebSocketSessionTests.restoreTlsConfiguration(); + } + + @Test + public void typedStringAndMappingSendsRejectInvalidJson() { + List messages = new CopyOnWriteArrayList<>(); + server = startServer(messages, new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), + new AtomicReference<>(), new AtomicReference<>(), false); + AgentsClientBuilder builder = builder(server.port()); + String raw = "{\"type\": \"response.create\"}"; + BinaryData mapping = BinaryData.fromObject(Collections.singletonMap("type", "response.cancel")); + BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() + .buildBetaVoiceAgentWebSocketAsyncClient() + .openWebSocketSession("agent", tlsOptions()) + .block(Duration.ofSeconds(5)); + try { + StepVerifier.create(session.sendEvent(BinaryData.fromString("not valid json"))) + .expectError(IllegalArgumentException.class) + .verify(Duration.ofSeconds(5)); + session.sendEvent(new RealtimeResponseCreateEvent()) + .then(session.sendEvent(BinaryData.fromString(raw))) + .then(session.sendEvent(mapping)) + .block(Duration.ofSeconds(5)); + StepVerifier.create(session.receiveEvents().take(3)).expectNextCount(3).verifyComplete(); + } finally { + session.close(); + } + assertEquals(3, messages.size()); + assertEquals("response.create", BinaryData.fromString(messages.get(0)).toObject(Map.class).get("type")); + assertEquals(raw, messages.get(1)); + assertEquals(mapping.toObject(Map.class), BinaryData.fromString(messages.get(2)).toObject(Map.class)); + } + + @Test + public void pingPongFramesAreNotApplicationEvents() { + Flux frames = Flux.defer(() -> Flux.just(new PingWebSocketFrame(), new PongWebSocketFrame(), + new TextWebSocketFrame("{\"type\":\"session.created\",\"session\":{}}"), + new TextWebSocketFrame("{\"type\":\"future.event\",\"foo\":\"bar\"}"))); + server = frameWebSocketServer(frames, false); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(Duration.ofSeconds(5)); + List events; + try { + events = session.receiveEvents().take(2).collectList().block(Duration.ofSeconds(5)); + } finally { + session.close(); + } + assertEquals(2, events.size()); + assertInstanceOf(RealtimeSessionCreatedEvent.class, events.get(0)); + RawRealtimeServerEvent unknown = assertInstanceOf(RawRealtimeServerEvent.class, events.get(1)); + assertEquals("bar", unknown.getRawEvent().toObject(Map.class).get("foo")); + } + + @Test + public void malformedEventsCanBeReportedAndSkipped() { + AtomicInteger failures = new AtomicInteger(); + server = frameWebSocketServer(Flux.defer(() -> Flux.just(new TextWebSocketFrame("{broken"), + new BinaryWebSocketFrame(Unpooled.wrappedBuffer(new byte[] { (byte) 0xc3, 0x28 })), + new BinaryWebSocketFrame(Unpooled.copiedBuffer(warningJson(), StandardCharsets.UTF_8)), + new TextWebSocketFrame(warningJson()))), true); + VoiceAgentWebSocketConnectionOptions options + = tlsOptions().setMalformedEventHandler(error -> failures.incrementAndGet()); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", options).block(Duration.ofSeconds(5)); + StepVerifier.create(session.receiveEvents()) + .assertNext(this::assertWarningEvent) + .assertNext(this::assertWarningEvent) + .verifyComplete(); + session.close(); + assertEquals(2, failures.get()); + } + + @Test + public void boundedQueuesHonorOverflowPolicies() { + for (VoiceAgentWebSocketOverflowStrategy strategy : VoiceAgentWebSocketOverflowStrategy.values()) { + server = frameWebSocketServer(Flux.range(0, 4) + .map(index -> new TextWebSocketFrame("{\"type\":\"future.event\",\"index\":" + index + "}")), true); + VoiceAgentWebSocketConnectionOptions options + = tlsOptions().setReceiveBufferCapacity(2).setOverflowStrategy(strategy); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", options).block(Duration.ofSeconds(5)); + boolean overflowError = strategy == VoiceAgentWebSocketOverflowStrategy.ERROR; + List received = new ArrayList<>(); + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + while (session.isOpen()) { + Thread.yield(); + } + }); + if (overflowError) { + StepVerifier.create(session.receiveEvents()) + .expectNextCount(2) + .expectError(IllegalStateException.class) + .verify(); + } else { + received.addAll(session.receiveEvents().collectList().block(Duration.ofSeconds(5))); + } + session.close(); + if (!overflowError) { + assertEquals(2, received.size()); + int first = strategy == VoiceAgentWebSocketOverflowStrategy.DROP_OLDEST ? 2 : 0; + for (int index = 0; index < received.size(); index++) { + assertEquals(first + index, + ((RawRealtimeServerEvent) received.get(index)).getRawEvent().toObject(Map.class).get("index")); + } + } + server.disposeNow(); + } + } + + @Test + public void messageSizeLimitCannotBeBypassedByRecoveryHandler() { + server = oneShotWebSocketServer(warningJson()); + AtomicBoolean recovered = new AtomicBoolean(); + VoiceAgentWebSocketConnectionOptions options + = tlsOptions().setMaxMessageSize(16).setMalformedEventHandler(error -> recovered.set(true)); + StepVerifier.create(createAsyncClient(server.port()).openWebSocketSession("agent", options) + .flatMapMany(BetaVoiceAgentWebSocketSessionAsyncClient::receiveEvents)).expectError().verify(); + assertFalse(recovered.get()); + } + + @Test + public void rawEventRoundTripsAndOptionsValidateBounds() throws Exception { + BinaryData payload = BinaryData.fromString("{\"type\":\"future.event\",\"nested\":{\"value\":42}}"); + RawRealtimeServerEvent event = new RawRealtimeServerEvent(payload); + RawRealtimeServerEvent copy = BinaryData.fromObject(event).toObject(RawRealtimeServerEvent.class); + assertEquals(payload.toObject(Map.class), copy.getRawEvent().toObject(Map.class)); + server = oneShotWebSocketServer("{\"type\":42,\"value\":1}"); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(Duration.ofSeconds(5)); + try { + StepVerifier.create(session.receiveEvents()) + .assertNext(received -> assertInstanceOf(RawRealtimeServerEvent.class, received)) + .verifyComplete(); + } finally { + session.close(); + } + VoiceAgentWebSocketConnectionOptions options = new VoiceAgentWebSocketConnectionOptions(); + assertThrows(IllegalArgumentException.class, () -> options.setReceiveBufferCapacity(0)); + assertThrows(IllegalArgumentException.class, () -> options.setReceiveBufferCapacity(65537)); + assertThrows(IllegalArgumentException.class, () -> options.setMaxMessageSize(0)); + assertThrows(NullPointerException.class, () -> options.setOverflowStrategy(null)); + } + + @Test + public void insecureEndpointsAreRejectedBeforeAuthentication() { + AtomicBoolean requested = new AtomicBoolean(); + TokenCredential credential = context -> { + requested.set(true); + return Mono.error(new AssertionError("Token retrieval must not run.")); + }; + for (String endpoint : new String[] { + "http://example.com", + "ws://example.com", + "https://user@example.com", + "https://example.com/#fragment" }) { + AgentsClientBuilder clientBuilder = new AgentsClientBuilder().endpoint(endpoint).credential(credential); + StepVerifier + .create(clientBuilder.beta().buildBetaVoiceAgentWebSocketAsyncClient().openWebSocketSession("agent")) + .expectError(IllegalArgumentException.class) + .verify(); + } + assertFalse(requested.get()); + } + + @Test + public void rawEventsUseCustomizedTlsTransport() { + List messages = new CopyOnWriteArrayList<>(); + server = VoiceAgentWebSocketSessionTests.tlsServer() + .host("localhost") + .port(0) + .handle((request, response) -> response.sendWebsocket( + (inbound, outbound) -> outbound.sendString(inbound.receive().asString().doOnNext(messages::add)).then(), + WebsocketServerSpec.builder().protocols("realtime").build())) + .bindNow(); + BinaryData payload = BinaryData.fromString("{\"type\":\"future.event\",\"value\":42}"); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(Duration.ofSeconds(5)); + StepVerifier.create(session.sendEvent(BinaryData.fromString("[]"))) + .expectError(IllegalArgumentException.class) + .verify(); + StepVerifier.create(session.receiveEvents().take(1)) + .then(() -> session.sendEvent(payload).block(Duration.ofSeconds(5))) + .assertNext(received -> assertEquals(42, + ((RawRealtimeServerEvent) received).getRawEvent().toObject(Map.class).get("value"))) + .verifyComplete(); + session.close(); + assertEquals(1, messages.size()); + } + + @Test + public void customCloseFrame() { + server = startServer(new CopyOnWriteArrayList<>(), new AtomicReference<>(), new AtomicReference<>(), + new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), false); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(Duration.ofSeconds(5)); + assertNotNull(session); + StepVerifier.create(session.closeAsync(1006, "invalid")).expectError(IllegalArgumentException.class).verify(); + session.closeAsync(4002, "done").block(Duration.ofSeconds(5)); + assertEquals(4002, session.getCloseCode()); + assertEquals("done", session.getCloseReason()); + } + + @Test + public void asyncSessionNegotiatesHandshakeAndExchangesTypedEvents() { + List clientMessages = new CopyOnWriteArrayList<>(); + AtomicReference requestUri = new AtomicReference<>(); + AtomicReference authorization = new AtomicReference<>(); + AtomicReference foundryFeatures = new AtomicReference<>(); + AtomicReference userAgent = new AtomicReference<>(); + AtomicReference customHeader = new AtomicReference<>(); + server + = startServer(clientMessages, requestUri, authorization, foundryFeatures, userAgent, customHeader, false); + AtomicReference> requestedScopes = new AtomicReference<>(); + TokenCredential credential = request -> { + requestedScopes.set(request.getScopes()); + return Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1))); + }; + VoiceAgentWebSocketConnectionOptions options + = tlsOptions().setStoreEnabled(true).setAgentVersionOverride("version 2"); + BetaVoiceAgentWebSocketAsyncClient client + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") + .credential(credential) + .configuration(Configuration.NONE) + .clientOptions(new ClientOptions().setApplicationId("test-app") + .setHeaders(Collections.singletonList(new Header("X-Test-Header", "test-value")))) + .beta() + .buildBetaVoiceAgentWebSocketAsyncClient(); + Mono sessionMono + = client.openWebSocketSession("agent name", options); + options.setStoreEnabled(false).setAgentVersionOverride("mutated"); + BetaVoiceAgentWebSocketSessionAsyncClient session = sessionMono.block(); + assertTrue(session.isOpen()); + StepVerifier.create(session.receiveEvents().take(4)) + .then(() -> session.sendText("hello").block()) + .assertNext(this::assertWarningEvent) + .then(() -> session.appendInputAudio(BinaryData.fromBytes(new byte[] { 1, 2, 3 })).block()) + .assertNext(this::assertWarningEvent) + .then(() -> session.createResponse().block()) + .assertNext(this::assertWarningEvent) + .then(() -> session.cancelResponse("response-1").block()) + .assertNext(this::assertWarningEvent) + .verifyComplete(); + assertEquals(Collections.singletonList("https://ai.azure.com/.default"), requestedScopes.get()); + assertEquals("Bearer test-token", authorization.get()); + assertEquals("VoiceAgents=V1Preview", foundryFeatures.get()); + assertTrue(userAgent.get().startsWith("test-app azsdk-java-")); + assertEquals("test-value", customHeader.get()); + String decodedUri = decode(requestUri.get()); + assertTrue(decodedUri.contains("/agents/agent name/endpoint/protocols/voice")); + assertTrue(decodedUri.contains("api-version=v1")); + assertTrue(decodedUri.contains("transport=websocket")); + assertTrue(decodedUri.contains("store=true")); + assertTrue(decodedUri.contains("x-agent-version-override=version 2")); + assertTrue(decodedUri.contains("x-ms-client-sdk=test-app azsdk-java-")); + assertEquals(4, clientMessages.size()); + StepVerifier.create(session.receiveEvents()) + .expectErrorMatches( + error -> error instanceof IllegalStateException && error.getMessage().contains("Only one")) + .verify(); + session.close(); + assertFalse(session.isOpen()); + } + + @Test + public void tokenFailureOccursBeforeNetworkAccess() { + AtomicBoolean connected = new AtomicBoolean(); + server = VoiceAgentWebSocketSessionTests.tlsServer().host("localhost").port(0).handle((request, response) -> { + connected.set(true); + return response.send(); + }).bindNow(); + TokenCredential credential = request -> Mono.error(new IllegalStateException("token unavailable")); + BetaVoiceAgentWebSocketAsyncClient client + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") + .credential(credential) + .configuration(Configuration.NONE) + .beta() + .buildBetaVoiceAgentWebSocketAsyncClient(); + StepVerifier.create(client.openWebSocketSession("agent", tlsOptions())) + .expectErrorMatches( + error -> error instanceof IllegalStateException && error.getMessage().contains("token unavailable")) + .verify(); + assertFalse(connected.get()); + } + + @Test + public void tokenAcquisitionDoesNotUseHandshakeTimeout() { + server = startServer(new CopyOnWriteArrayList<>(), new AtomicReference<>(), new AtomicReference<>(), + new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), false); + TokenCredential credential = request -> Mono.delay(Duration.ofMillis(1500)) + .map(ignored -> new AccessToken("test-token", OffsetDateTime.now().plusHours(1))); + BetaVoiceAgentWebSocketAsyncClient client + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") + .credential(credential) + .configuration(Configuration.NONE) + .beta() + .buildBetaVoiceAgentWebSocketAsyncClient(); + VoiceAgentWebSocketConnectionOptions options = tlsOptions().setHandshakeTimeout(Duration.ofSeconds(1)); + StepVerifier.withVirtualTime(() -> client.openWebSocketSession("agent", options).flatMap(session -> { + assertTrue(session.isOpen()); + return session.closeAsync(); + })).thenAwait(Duration.ofMillis(1500)).verifyComplete(); + } + + @Test + public void rejectedHandshakeMapsConflictToAzureException() { + server = VoiceAgentWebSocketSessionTests.tlsServer() + .host("localhost") + .port(0) + .handle( + (request, response) -> response.status(HttpResponseStatus.CONFLICT).sendString(Mono.just("conflict"))) + .bindNow(); + StepVerifier.create(createAsyncClient(server.port()).openWebSocketSession("disabled-agent", tlsOptions())) + .expectErrorSatisfies(error -> { + ResourceModifiedException exception = assertInstanceOf(ResourceModifiedException.class, error); + assertEquals(409, exception.getResponse().getStatusCode()); + }) + .verify(); + } + + @Test + public void nettyHandshakeResponseExposesBufferedBody() { + DefaultFullHttpResponse nettyResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, + HttpResponseStatus.CONFLICT, Unpooled.copiedBuffer("conflict", StandardCharsets.UTF_8)); + VoiceAgentWebSocketHttpResponse response + = new VoiceAgentWebSocketHttpResponse(URI.create("wss://example.com"), nettyResponse); + assertEquals("conflict", response.getBodyAsString().block()); + } + + @Test + public void cancellingAsyncConnectCancelsTokenRequest() { + AtomicBoolean tokenRequestCancelled = new AtomicBoolean(); + TokenCredential credential + = request -> Mono.never().doOnCancel(() -> tokenRequestCancelled.set(true)); + BetaVoiceAgentWebSocketAsyncClient client = new AgentsClientBuilder().endpoint("https://example.com") + .credential(credential) + .configuration(Configuration.NONE) + .beta() + .buildBetaVoiceAgentWebSocketAsyncClient(); + StepVerifier.create(client.openWebSocketSession("agent", tlsOptions())).thenCancel().verify(); + assertTrue(tokenRequestCancelled.get()); + } + + @Test + public void unknownEventFallsBackToRealtimeServerEvent() { + server = oneShotWebSocketServer("{\"type\":\"future.event\",\"value\":42}"); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); + StepVerifier.create(session.receiveEvents()).assertNext(event -> { + assertEquals("future.event", event.getType().toString()); + RawRealtimeServerEvent raw = assertInstanceOf(RawRealtimeServerEvent.class, event); + assertEquals(42, raw.getRawEvent().toObject(Map.class).get("value")); + }).verifyComplete(); + session.close(); + } + + @Test + public void fragmentedTextFrameIsAggregated() { + String message = warningJson(); + int split = message.length() / 2; + server = frameWebSocketServer(Flux.just(new TextWebSocketFrame(false, 0, message.substring(0, split)), + new ContinuationWebSocketFrame(true, 0, message.substring(split))), true); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); + StepVerifier.create(session.receiveEvents()).assertNext(this::assertWarningEvent).verifyComplete(); + session.close(); + } + + @Test + public void binaryJsonFrameIsParsed() { + server = frameWebSocketServer( + Mono.just(new BinaryWebSocketFrame(Unpooled.copiedBuffer(warningJson(), StandardCharsets.UTF_8))), true); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); + StepVerifier.create(session.receiveEvents()).assertNext(this::assertWarningEvent).verifyComplete(); + session.close(); + } + + @Test + public void malformedJsonTerminatesReceiveStream() { + server = oneShotWebSocketServer("{not-json"); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); + StepVerifier.create(session.receiveEvents()).expectError().verify(); + assertFalse(session.isOpen()); + } + + @Test + public void closeIsIdempotentAndSendAfterCloseFails() { + server = startServer(new CopyOnWriteArrayList<>(), new AtomicReference<>(), new AtomicReference<>(), + new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), false); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); + StepVerifier.create(session.closeAsync().then(session.closeAsync())).verifyComplete(); + StepVerifier.create(session.sendText("after close")) + .expectErrorMatches( + error -> error instanceof IllegalStateException && error.getMessage().contains("not open")) + .verify(); + } + + @Test + public void secureSessionUsesWssAndReceivesTypedEvent() throws Exception { + server = oneShotWebSocketServer(warningJson()); + BetaVoiceAgentWebSocketSessionAsyncClient session + = createAsyncClient(server.port()).openWebSocketSession("secure-agent", tlsOptions()) + .block(Duration.ofSeconds(5)); + assertEquals("wss", session.getEndpoint().getScheme()); + StepVerifier.create(session.receiveEvents()).assertNext(this::assertWarningEvent).verifyComplete(); + session.close(); + } + + private AgentsClientBuilder builder(int port) { + return new AgentsClientBuilder().endpoint("https://localhost:" + port) + .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))); + } + + private BetaVoiceAgentWebSocketAsyncClient createAsyncClient(int port) { + return builder(port).configuration(Configuration.NONE).beta().buildBetaVoiceAgentWebSocketAsyncClient(); + } + + private static VoiceAgentWebSocketConnectionOptions tlsOptions() { + return new VoiceAgentWebSocketConnectionOptions(); + } + + private DisposableServer frameWebSocketServer(org.reactivestreams.Publisher frames, + boolean close) { + WebsocketServerSpec spec = WebsocketServerSpec.builder().protocols("realtime").build(); + return VoiceAgentWebSocketSessionTests.tlsServer() + .host("localhost") + .port(0) + .handle((request, + response) -> response.sendWebsocket((inbound, outbound) -> close + ? outbound.sendObject(frames).then(outbound.sendClose()) + : outbound.sendObject(frames).then(inbound.receive().then()), spec)) + .bindNow(); + } + + private DisposableServer oneShotWebSocketServer(String message) { + WebsocketServerSpec spec = WebsocketServerSpec.builder().protocols("realtime").build(); + return VoiceAgentWebSocketSessionTests.tlsServer() + .host("localhost") + .port(0) + .handle((request, response) -> response.sendWebsocket((inbound, + outbound) -> outbound.sendString(Mono.just(message), StandardCharsets.UTF_8).then(outbound.sendClose()), + spec)) + .bindNow(); + } + + private DisposableServer startServer(List clientMessages, AtomicReference requestUri, + AtomicReference authorization, AtomicReference foundryFeatures, + AtomicReference userAgent, AtomicReference customHeader, boolean sendInitialEvent) { + WebsocketServerSpec spec = WebsocketServerSpec.builder().protocols("realtime").build(); + return VoiceAgentWebSocketSessionTests.tlsServer().host("localhost").port(0).handle((request, response) -> { + requestUri.set(request.uri()); + authorization.set(request.requestHeaders().get(HttpHeaderNames.AUTHORIZATION)); + foundryFeatures.set(request.requestHeaders().get("Foundry-Features")); + userAgent.set(request.requestHeaders().get(HttpHeaderNames.USER_AGENT)); + customHeader.set(request.requestHeaders().get("X-Test-Header")); + return response.sendWebsocket((inbound, outbound) -> { + Flux replies = inbound.receive() + .asString(StandardCharsets.UTF_8) + .doOnNext(clientMessages::add) + .map(ignored -> warningJson()); + if (sendInitialEvent) { + replies = replies.startWith(warningJson()); + } + return outbound.sendString(replies, StandardCharsets.UTF_8).then(); + }, spec); + }).bindNow(); + } + + private void assertWarningEvent(RealtimeServerEvent event) { + VoiceAgentWarningEvent warning = assertInstanceOf(VoiceAgentWarningEvent.class, event); + assertEquals("loopback warning", warning.getWarning().getMessage()); + assertEquals("test_warning", warning.getWarning().getCode()); + } + + private static String warningJson() { + return "{\"type\":\"warning\",\"event_id\":\"event-1\",\"warning\":{" + + "\"message\":\"loopback warning\",\"code\":\"test_warning\"}}"; + } + + private static String decode(String value) { + try { + return URLDecoder.decode(value, StandardCharsets.UTF_8.name()); + } catch (Exception error) { + throw new IllegalStateException(error); + } + } +} diff --git a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionTests.java b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionTests.java index b6d8e4ec31c3..da6b1f6dafbd 100644 --- a/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionTests.java +++ b/sdk/ai/azure-ai-agents/src/test/java/com/azure/ai/agents/voice/VoiceAgentWebSocketSessionTests.java @@ -4,39 +4,32 @@ package com.azure.ai.agents.voice; import com.azure.ai.agents.AgentsClientBuilder; -import com.azure.ai.agents.BetaVoiceAgentWebSocketAsyncClient; import com.azure.ai.agents.BetaVoiceAgentWebSocketClient; -import com.azure.ai.agents.BetaVoiceAgentWebSocketSessionAsyncClient; import com.azure.ai.agents.BetaVoiceAgentWebSocketSessionClient; -import com.azure.ai.agents.implementation.realtime.VoiceAgentWebSocketHttpResponse; import com.azure.ai.agents.models.RawRealtimeServerEvent; import com.azure.ai.agents.models.RealtimeResponseCreateEvent; import com.azure.ai.agents.models.RealtimeSessionCreatedEvent; import com.azure.ai.agents.models.RealtimeServerEvent; import com.azure.ai.agents.models.VoiceAgentWarningEvent; -import com.azure.ai.agents.models.VoiceAgentTransport; import com.azure.ai.agents.models.VoiceAgentWebSocketConnectionOptions; import com.azure.ai.agents.models.VoiceAgentWebSocketOverflowStrategy; import com.azure.core.credential.AccessToken; import com.azure.core.credential.TokenCredential; import com.azure.core.exception.ResourceModifiedException; +import com.azure.core.http.policy.HttpLogOptions; import com.azure.core.util.BinaryData; import com.azure.core.util.ClientOptions; import com.azure.core.util.Configuration; import com.azure.core.util.Header; import io.netty.buffer.Unpooled; -import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.HttpHeaderNames; import io.netty.handler.codec.http.HttpResponseStatus; -import io.netty.handler.codec.http.HttpVersion; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; -import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import java.io.InputStream; -import java.net.URI; import java.net.URLDecoder; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -46,9 +39,9 @@ import java.time.Duration; import java.time.OffsetDateTime; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Iterator; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; @@ -62,25 +55,24 @@ import javax.net.ssl.TrustManagerFactory; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.api.parallel.ResourceLock; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.netty.DisposableServer; import reactor.netty.http.Http11SslContextSpec; import reactor.netty.http.server.HttpServer; import reactor.netty.http.server.WebsocketServerSpec; -import reactor.test.StepVerifier; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; -import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; +@ResourceLock("voice-agent-websocket-tls") public class VoiceAgentWebSocketSessionTests { private static final String TRUST_STORE_PROPERTY = "javax.net.ssl.trustStore"; private static final String TRUST_STORE_PASSWORD_PROPERTY = "javax.net.ssl.trustStorePassword"; @@ -91,15 +83,15 @@ public class VoiceAgentWebSocketSessionTests { private static final SSLContext ORIGINAL_SSL_CONTEXT = getDefaultSslContext(); private static final TestCertificate TLS_CERTIFICATE = TestCertificate.create(); - static { + @BeforeAll + static void installTlsCertificate() { TLS_CERTIFICATE.installTrustStore(); } private DisposableServer server; - @ParameterizedTest - @ValueSource(booleans = { false, true }) - public void handshakeOverridesPreserveQueryAndSingleUserAgent(boolean async) { + @Test + public void handshakeUsesSdkManagedProtocolValues() { AtomicReference requestUri = new AtomicReference<>(); AtomicReference headers = new AtomicReference<>(); server = tlsServer().host("localhost").port(0).handle((request, response) -> { @@ -111,58 +103,62 @@ public void handshakeOverridesPreserveQueryAndSingleUserAgent(boolean async) { AgentsClientBuilder builder = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project/") .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))) - .configuration(Configuration.NONE); - for (String userAgentHeader : new String[] { "", "User-Agent", "user-agent" }) { - Map extra = new LinkedHashMap<>(); - extra.put("X-Custom", "custom-value"); - extra.put("Authorization", "must-not-override-token"); - extra.put("Sec-WebSocket-Protocol", "other"); - VoiceAgentWebSocketConnectionOptions options - = tlsOptions().setExtraQuery(Collections.singletonMap("foo", "bar value")); - if (!userAgentHeader.isEmpty()) { - extra.put(userAgentHeader, "custom-user-agent"); - options.setConnectionUrl(URI.create("wss://localhost:" + server.port() + "/custom?sig=abc")); - } - options.setExtraHeaders(extra); - if (async) { - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent name", options) - .block(Duration.ofSeconds(5)); - session.closeAsync().block(Duration.ofSeconds(5)); - assertFalse(session.isOpen()); - } else { - BetaVoiceAgentWebSocketSessionClient session - = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent name", options); - session.close(); - assertFalse(session.isOpen()); - } - assertEquals("Bearer test-token", headers.get().get(HttpHeaderNames.AUTHORIZATION)); - assertEquals("realtime", headers.get().get(HttpHeaderNames.SEC_WEBSOCKET_PROTOCOL)); - assertEquals("VoiceAgents=V1Preview", headers.get().get("Foundry-Features")); - assertEquals("custom-value", headers.get().get("X-Custom")); - assertEquals(1, headers.get().getAll(HttpHeaderNames.USER_AGENT).size()); - String userAgent = headers.get().get(HttpHeaderNames.USER_AGENT); - String uri = decode(requestUri.get()); - assertTrue(uri.contains("api-version=v1")); - assertTrue(uri.contains("foo=bar value")); - assertEquals(1, requestUri.get().chars().filter(character -> character == '?').count()); - if (userAgentHeader.isEmpty()) { - assertTrue(userAgent.startsWith("azsdk-java-azure-ai-agents/"), userAgent); - assertTrue(uri.contains("x-ms-client-sdk=" + userAgent)); - assertTrue(uri.startsWith("/api/projects/project/agents/agent name/endpoint/protocols/voice?")); - } else { - assertEquals("custom-user-agent", userAgent); - assertTrue(uri.startsWith("/custom?")); - assertTrue(uri.contains("sig=abc")); - assertTrue(uri.contains("x-ms-client-sdk=azsdk-java-azure-ai-agents/"), uri); - } - } + .configuration(Configuration.NONE) + .httpLogOptions(new HttpLogOptions().setApplicationId("log-app")) + .clientOptions(new ClientOptions().setApplicationId("client-app") + .setHeaders(Arrays.asList(new Header("X-Custom", "custom-value"), + new Header("Authorization", "Basic override"), new Header("User-Agent", "override-agent"), + new Header("Foundry-Features", "override-feature"), + new Header("Sec-WebSocket-Protocol", "override-protocol")))); + + BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent name"); + session.close(); + assertFalse(session.isOpen()); + + assertEquals("Bearer test-token", headers.get().get(HttpHeaderNames.AUTHORIZATION)); + assertEquals("realtime", headers.get().get(HttpHeaderNames.SEC_WEBSOCKET_PROTOCOL)); + assertEquals("VoiceAgents=V1Preview", headers.get().get("Foundry-Features")); + assertEquals("custom-value", headers.get().get("X-Custom")); + assertEquals(1, headers.get().getAll(HttpHeaderNames.USER_AGENT).size()); + String userAgent = headers.get().get(HttpHeaderNames.USER_AGENT); + assertTrue(userAgent.startsWith("client-app azsdk-java-azure-ai-agents/"), userAgent); + assertFalse(userAgent.contains("log-app"), userAgent); + assertFalse(userAgent.contains("override-agent"), userAgent); + String uri = decode(requestUri.get()); + assertTrue(uri.contains("api-version=v1")); + assertTrue(uri.contains("transport=websocket")); + assertTrue(uri.contains("x-ms-client-sdk=" + userAgent)); + assertTrue(uri.startsWith("/api/projects/project/agents/agent name/endpoint/protocols/voice?")); } - @ParameterizedTest - @ValueSource(booleans = { false, true }) - public void typedStringAndMappingSendsRejectInvalidJson(boolean async) { + @Test + public void handshakeUsesHttpLogApplicationIdFallback() { + AtomicReference requestUri = new AtomicReference<>(); + AtomicReference userAgent = new AtomicReference<>(); + server = tlsServer().host("localhost").port(0).handle((request, response) -> { + requestUri.set(request.uri()); + userAgent.set(request.requestHeaders().get(HttpHeaderNames.USER_AGENT)); + return response.sendWebsocket((inbound, outbound) -> inbound.receive().then(), + WebsocketServerSpec.builder().protocols("realtime").build()); + }).bindNow(); + AgentsClientBuilder builder + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") + .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))) + .configuration(Configuration.NONE) + .httpLogOptions(new HttpLogOptions().setApplicationId("log-app")) + .clientOptions(new ClientOptions()); + + BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent"); + session.close(); + + assertTrue(userAgent.get().startsWith("log-app azsdk-java-azure-ai-agents/"), userAgent.get()); + assertTrue(decode(requestUri.get()).contains("x-ms-client-sdk=" + userAgent.get())); + } + + @Test + public void typedStringAndMappingSendsRejectInvalidJson() { List messages = new CopyOnWriteArrayList<>(); server = startServer(messages, new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), false); @@ -170,35 +166,16 @@ public void typedStringAndMappingSendsRejectInvalidJson(boolean async) { .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))); String raw = "{\"type\": \"response.create\"}"; BinaryData mapping = BinaryData.fromObject(Collections.singletonMap("type", "response.cancel")); - if (async) { - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", tlsOptions()) - .block(Duration.ofSeconds(5)); - try { - StepVerifier.create(session.sendEvent(BinaryData.fromString("not valid json"))) - .expectError(IllegalArgumentException.class) - .verify(Duration.ofSeconds(5)); - session.sendEvent(new RealtimeResponseCreateEvent()) - .then(session.sendEvent(BinaryData.fromString(raw))) - .then(session.sendEvent(mapping)) - .block(Duration.ofSeconds(5)); - StepVerifier.create(session.receiveEvents().take(3)).expectNextCount(3).verifyComplete(); - } finally { - session.close(); - } - } else { - try (BetaVoiceAgentWebSocketSessionClient session - = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", tlsOptions())) { - assertThrows(IllegalArgumentException.class, - () -> session.sendEvent(BinaryData.fromString("not valid json"))); - session.sendEvent(new RealtimeResponseCreateEvent()); - session.sendEvent(BinaryData.fromString(raw)); - session.sendEvent(mapping); - Iterator events = session.receiveEvents(Duration.ofSeconds(5)).iterator(); - for (int index = 0; index < 3; index++) { - assertWarningEvent(events.next()); - } + try (BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", tlsOptions())) { + assertThrows(IllegalArgumentException.class, + () -> session.sendEvent(BinaryData.fromString("not valid json"))); + session.sendEvent(new RealtimeResponseCreateEvent()); + session.sendEvent(BinaryData.fromString(raw)); + session.sendEvent(mapping); + Iterator events = session.receiveEvents(Duration.ofSeconds(5)).iterator(); + for (int index = 0; index < 3; index++) { + assertWarningEvent(events.next()); } } assertEquals(3, messages.size()); @@ -207,9 +184,8 @@ public void typedStringAndMappingSendsRejectInvalidJson(boolean async) { assertEquals(mapping.toObject(Map.class), BinaryData.fromString(messages.get(2)).toObject(Map.class)); } - @ParameterizedTest - @ValueSource(booleans = { false, true }) - public void pingPongFramesAreNotApplicationEvents(boolean async) { + @Test + public void pingPongFramesAreNotApplicationEvents() { Flux frames = Flux.defer(() -> Flux.just(new PingWebSocketFrame(), new PongWebSocketFrame(), new TextWebSocketFrame("{\"type\":\"session.created\",\"session\":{}}"), new TextWebSocketFrame("{\"type\":\"future.event\",\"foo\":\"bar\"}"))); @@ -222,23 +198,11 @@ public void pingPongFramesAreNotApplicationEvents(boolean async) { AgentsClientBuilder builder = new AgentsClientBuilder().endpoint("https://localhost:" + server.port()) .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))); List events = new ArrayList<>(); - if (async) { - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", tlsOptions()) - .block(Duration.ofSeconds(5)); - try { - events.addAll(session.receiveEvents().take(2).collectList().block(Duration.ofSeconds(5))); - } finally { - session.close(); - } - } else { - try (BetaVoiceAgentWebSocketSessionClient session - = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", tlsOptions())) { - Iterator iterator = session.receiveEvents(Duration.ofSeconds(5)).iterator(); - events.add(iterator.next()); - events.add(iterator.next()); - } + try (BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", tlsOptions())) { + Iterator iterator = session.receiveEvents(Duration.ofSeconds(5)).iterator(); + events.add(iterator.next()); + events.add(iterator.next()); } assertEquals(2, events.size()); assertInstanceOf(RealtimeSessionCreatedEvent.class, events.get(0)); @@ -247,28 +211,7 @@ public void pingPongFramesAreNotApplicationEvents(boolean async) { } @Test - public void explicitDefaultPortOverrideIsTrustedBeforeAuthentication() { - AtomicInteger tokens = new AtomicInteger(); - IllegalStateException tokenError = new IllegalStateException("Stop before network access."); - AgentsClientBuilder builder - = new AgentsClientBuilder().endpoint("https://example.com/api/projects/project").credential(request -> { - tokens.incrementAndGet(); - return Mono.error(tokenError); - }); - VoiceAgentWebSocketConnectionOptions options - = new VoiceAgentWebSocketConnectionOptions().setConnectionUrl(URI.create("wss://example.com:443/custom")); - assertEquals(tokenError, assertThrows(IllegalStateException.class, - () -> builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options))); - StepVerifier - .create(builder.beta().buildBetaVoiceAgentWebSocketAsyncClient().openWebSocketSession("agent", options)) - .expectErrorSatisfies(error -> assertEquals(tokenError, error)) - .verify(Duration.ofSeconds(5)); - assertEquals(2, tokens.get()); - } - - @ParameterizedTest - @ValueSource(booleans = { false, true }) - public void malformedEventsCanBeReportedAndSkipped(boolean async) { + public void malformedEventsCanBeReportedAndSkipped() { AtomicInteger failures = new AtomicInteger(); server = frameWebSocketServer(Flux.defer(() -> Flux.just(new TextWebSocketFrame("{broken"), new BinaryWebSocketFrame(Unpooled.wrappedBuffer(new byte[] { (byte) 0xc3, 0x28 })), @@ -278,31 +221,18 @@ public void malformedEventsCanBeReportedAndSkipped(boolean async) { = tlsOptions().setMalformedEventHandler(error -> failures.incrementAndGet()); AgentsClientBuilder builder = new AgentsClientBuilder().endpoint("https://localhost:" + server.port()) .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))); - if (async) { - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", options) - .block(Duration.ofSeconds(5)); - StepVerifier.create(session.receiveEvents()) - .assertNext(this::assertWarningEvent) - .assertNext(this::assertWarningEvent) - .verifyComplete(); - session.close(); - } else { - try (BetaVoiceAgentWebSocketSessionClient session - = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)) { - Iterator events = session.receiveEvents(Duration.ofSeconds(5)).iterator(); - assertWarningEvent(events.next()); - assertWarningEvent(events.next()); - assertFalse(events.hasNext()); - } + try (BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)) { + Iterator events = session.receiveEvents(Duration.ofSeconds(5)).iterator(); + assertWarningEvent(events.next()); + assertWarningEvent(events.next()); + assertFalse(events.hasNext()); } assertEquals(2, failures.get()); } - @ParameterizedTest - @ValueSource(booleans = { false, true }) - public void boundedQueuesHonorOverflowPolicies(boolean async) { + @Test + public void boundedQueuesHonorOverflowPolicies() { for (VoiceAgentWebSocketOverflowStrategy strategy : VoiceAgentWebSocketOverflowStrategy.values()) { server = frameWebSocketServer(Flux.range(0, 4) .map(index -> new TextWebSocketFrame("{\"type\":\"future.event\",\"index\":" + index + "}"))); @@ -312,39 +242,18 @@ public void boundedQueuesHonorOverflowPolicies(boolean async) { .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))); List received = new ArrayList<>(); boolean overflowError = strategy == VoiceAgentWebSocketOverflowStrategy.ERROR; - if (async) { - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", options) - .block(Duration.ofSeconds(5)); + try (BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)) { assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { while (session.isOpen()) { Thread.yield(); } }); + Iterator events = session.receiveEvents(Duration.ofSeconds(5)).iterator(); if (overflowError) { - StepVerifier.create(session.receiveEvents()) - .expectNextCount(2) - .expectError(IllegalStateException.class) - .verify(); + assertThrows(IllegalStateException.class, events::hasNext); } else { - received.addAll(session.receiveEvents().collectList().block(Duration.ofSeconds(5))); - } - session.close(); - } else { - try (BetaVoiceAgentWebSocketSessionClient session - = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)) { - assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { - while (session.isOpen()) { - Thread.yield(); - } - }); - Iterator events = session.receiveEvents(Duration.ofSeconds(5)).iterator(); - if (overflowError) { - assertThrows(IllegalStateException.class, events::hasNext); - } else { - events.forEachRemaining(received::add); - } + events.forEachRemaining(received::add); } } if (!overflowError) { @@ -359,54 +268,23 @@ public void boundedQueuesHonorOverflowPolicies(boolean async) { } } - @ParameterizedTest - @ValueSource(booleans = { false, true }) - public void messageSizeLimitCannotBeBypassedByRecoveryHandler(boolean async) { + @Test + public void messageSizeLimitCannotBeBypassedByRecoveryHandler() { server = oneShotWebSocketServer(warningJson()); AtomicBoolean recovered = new AtomicBoolean(); VoiceAgentWebSocketConnectionOptions options = tlsOptions().setMaxMessageSize(16).setMalformedEventHandler(error -> recovered.set(true)); AgentsClientBuilder builder = new AgentsClientBuilder().endpoint("https://localhost:" + server.port()) .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))); - if (async) { - StepVerifier.create(builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", options) - .flatMapMany(BetaVoiceAgentWebSocketSessionAsyncClient::receiveEvents)).expectError().verify(); - } else { - assertThrows(RuntimeException.class, () -> { - try (BetaVoiceAgentWebSocketSessionClient session - = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)) { - session.receiveEvents(Duration.ofSeconds(5)).iterator().next(); - } - }); - } + assertThrows(RuntimeException.class, () -> { + try (BetaVoiceAgentWebSocketSessionClient session + = builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)) { + session.receiveEvents(Duration.ofSeconds(5)).iterator().next(); + } + }); assertFalse(recovered.get()); } - @Test - public void rawEventRoundTripsAndOptionsValidateBounds() throws Exception { - BinaryData payload = BinaryData.fromString("{\"type\":\"future.event\",\"nested\":{\"value\":42}}"); - RawRealtimeServerEvent event = new RawRealtimeServerEvent(payload); - RawRealtimeServerEvent copy = BinaryData.fromObject(event).toObject(RawRealtimeServerEvent.class); - assertEquals(payload.toObject(Map.class), copy.getRawEvent().toObject(Map.class)); - server = oneShotWebSocketServer("{\"type\":42,\"value\":1}"); - BetaVoiceAgentWebSocketSessionAsyncClient session - = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(Duration.ofSeconds(5)); - try { - StepVerifier.create(session.receiveEvents()) - .assertNext(received -> assertInstanceOf(RawRealtimeServerEvent.class, received)) - .verifyComplete(); - } finally { - session.close(); - } - VoiceAgentWebSocketConnectionOptions options = new VoiceAgentWebSocketConnectionOptions(); - assertThrows(IllegalArgumentException.class, () -> options.setReceiveBufferCapacity(0)); - assertThrows(IllegalArgumentException.class, () -> options.setReceiveBufferCapacity(65537)); - assertThrows(IllegalArgumentException.class, () -> options.setMaxMessageSize(0)); - assertThrows(NullPointerException.class, () -> options.setOverflowStrategy(null)); - } - @Test public void insecureEndpointsAreRejectedBeforeAuthentication() { AtomicBoolean requested = new AtomicBoolean(); @@ -422,28 +300,11 @@ public void insecureEndpointsAreRejectedBeforeAuthentication() { AgentsClientBuilder builder = new AgentsClientBuilder().endpoint(endpoint).credential(credential); assertThrows(IllegalArgumentException.class, () -> builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent")); - StepVerifier.create(builder.beta().buildBetaVoiceAgentWebSocketAsyncClient().openWebSocketSession("agent")) - .expectError(IllegalArgumentException.class) - .verify(); - } - AgentsClientBuilder builder = new AgentsClientBuilder().endpoint("https://example.com").credential(credential); - for (String override : new String[] { - "ws://example.com", - "wss://other.example.com", - "wss://example.com:8443" }) { - VoiceAgentWebSocketConnectionOptions options - = new VoiceAgentWebSocketConnectionOptions().setConnectionUrl(URI.create(override)); - assertThrows(IllegalArgumentException.class, - () -> builder.beta().buildBetaVoiceAgentWebSocketClient().openWebSocketSession("agent", options)); - StepVerifier - .create(builder.beta().buildBetaVoiceAgentWebSocketAsyncClient().openWebSocketSession("agent", options)) - .expectError(IllegalArgumentException.class) - .verify(); } assertFalse(requested.get()); } - private static HttpServer tlsServer() { + static HttpServer tlsServer() { return HttpServer.create() .secure(ssl -> ssl.sslContext(Http11SslContextSpec.forServer(TLS_CERTIFICATE.keyManagerFactory))); } @@ -453,7 +314,25 @@ private static VoiceAgentWebSocketConnectionOptions tlsOptions() { } @Test - public void rawEventsUseCustomizedTlsTransports() { + public void syncReceiveStreamRejectsSecondIterator() { + server = startServer(new CopyOnWriteArrayList<>(), new AtomicReference<>(), new AtomicReference<>(), + new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), false); + BetaVoiceAgentWebSocketClient client + = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") + .credential(request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1)))) + .beta() + .buildBetaVoiceAgentWebSocketClient(); + + try (BetaVoiceAgentWebSocketSessionClient session = client.openWebSocketSession("agent", tlsOptions())) { + Iterable events = session.receiveEvents(Duration.ofSeconds(5)); + events.iterator(); + IllegalStateException exception = assertThrows(IllegalStateException.class, events::iterator); + assertEquals("The receiveEvents stream may only be iterated once.", exception.getMessage()); + } + } + + @Test + public void rawEventsUseCustomizedTlsTransport() { List messages = new CopyOnWriteArrayList<>(); server = tlsServer().host("localhost") .port(0) @@ -473,20 +352,7 @@ public void rawEventsUseCustomizedTlsTransports() { session.receiveEvents(Duration.ofSeconds(5)).iterator().next()); assertEquals(payload.toObject(Map.class), received.getRawEvent().toObject(Map.class)); } - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", tlsOptions()) - .block(Duration.ofSeconds(5)); - StepVerifier.create(session.sendEvent(BinaryData.fromString("[]"))) - .expectError(IllegalArgumentException.class) - .verify(); - StepVerifier.create(session.receiveEvents().take(1)) - .then(() -> session.sendEvent(payload).block(Duration.ofSeconds(5))) - .assertNext(event -> assertEquals(42, - ((RawRealtimeServerEvent) event).getRawEvent().toObject(Map.class).get("value"))) - .verifyComplete(); - session.close(); - assertEquals(2, messages.size()); + assertEquals(1, messages.size()); } @Test @@ -507,15 +373,6 @@ public void customCloseFrameAndReceiveTimeout() { assertEquals(4001, session.getCloseCode()); assertEquals("finished", session.getCloseReason()); } - BetaVoiceAgentWebSocketSessionAsyncClient session = builder.beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("agent", tlsOptions()) - .block(Duration.ofSeconds(5)); - assertNotNull(session); - StepVerifier.create(session.closeAsync(1006, "invalid")).expectError(IllegalArgumentException.class).verify(); - session.closeAsync(4002, "done").block(Duration.ofSeconds(5)); - assertEquals(4002, session.getCloseCode()); - assertEquals("done", session.getCloseReason()); } @AfterEach @@ -526,84 +383,11 @@ public void disposeServer() { } @AfterAll - public static void deleteTlsCertificate() { + static void restoreTlsConfiguration() { SSLContext.setDefault(ORIGINAL_SSL_CONTEXT); restoreProperty(TRUST_STORE_PROPERTY, ORIGINAL_TRUST_STORE); restoreProperty(TRUST_STORE_PASSWORD_PROPERTY, ORIGINAL_TRUST_STORE_PASSWORD); restoreProperty(TRUST_STORE_TYPE_PROPERTY, ORIGINAL_TRUST_STORE_TYPE); - TLS_CERTIFICATE.delete(); - } - - @Test - public void asyncSessionNegotiatesHandshakeAndExchangesTypedEvents() { - List clientMessages = new CopyOnWriteArrayList<>(); - AtomicReference requestUri = new AtomicReference<>(); - AtomicReference authorization = new AtomicReference<>(); - AtomicReference foundryFeatures = new AtomicReference<>(); - AtomicReference userAgent = new AtomicReference<>(); - AtomicReference customHeader = new AtomicReference<>(); - server - = startServer(clientMessages, requestUri, authorization, foundryFeatures, userAgent, customHeader, false); - AtomicReference> requestedScopes = new AtomicReference<>(); - TokenCredential credential = request -> { - requestedScopes.set(request.getScopes()); - return Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1))); - }; - VoiceAgentWebSocketConnectionOptions options = tlsOptions().setTransport(VoiceAgentTransport.WEBSOCKET) - .setStoreEnabled(true) - .setAgentVersionOverride("version 2"); - BetaVoiceAgentWebSocketAsyncClient client - = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") - .credential(credential) - .configuration(Configuration.NONE) - .clientOptions(new ClientOptions().setApplicationId("test-app") - .setHeaders(Collections.singletonList(new Header("X-Test-Header", "test-value")))) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient(); - - Mono sessionMono - = client.openWebSocketSession("agent name", options); - options.setStoreEnabled(false).setAgentVersionOverride("mutated"); - BetaVoiceAgentWebSocketSessionAsyncClient session = sessionMono.block(); - assertTrue(session.isOpen()); - - StepVerifier.create(session.receiveEvents().take(4)) - .then(() -> session.sendText("hello").block()) - .assertNext(this::assertWarningEvent) - .then(() -> session.appendInputAudio(BinaryData.fromBytes(new byte[] { 1, 2, 3 })).block()) - .assertNext(this::assertWarningEvent) - .then(() -> session.createResponse().block()) - .assertNext(this::assertWarningEvent) - .then(() -> session.cancelResponse("response-1").block()) - .assertNext(this::assertWarningEvent) - .verifyComplete(); - - assertEquals(Collections.singletonList("https://ai.azure.com/.default"), requestedScopes.get()); - assertEquals("Bearer test-token", authorization.get()); - assertEquals("VoiceAgents=V1Preview", foundryFeatures.get()); - assertTrue(userAgent.get().startsWith("test-app azsdk-java-")); - assertEquals("test-value", customHeader.get()); - String decodedUri = decode(requestUri.get()); - assertTrue(decodedUri.contains("/agents/agent name/endpoint/protocols/voice")); - assertTrue(decodedUri.contains("api-version=v1")); - assertTrue(decodedUri.contains("transport=websocket")); - assertTrue(decodedUri.contains("store=true")); - assertTrue(decodedUri.contains("x-agent-version-override=version 2")); - assertTrue(decodedUri.contains("x-ms-client-sdk=test-app azsdk-java-")); - assertEquals(4, clientMessages.size()); - assertTrue(clientMessages.get(0).contains("\"type\":\"conversation.item.create\"")); - assertTrue(clientMessages.get(0).contains("\"role\":\"user\"")); - assertTrue(clientMessages.get(0).contains("\"text\":\"hello\"")); - assertTrue(clientMessages.get(1).contains("\"audio\":\"AQID\"")); - assertTrue(clientMessages.get(2).contains("\"type\":\"response.create\"")); - assertTrue(clientMessages.get(3).contains("\"response_id\":\"response-1\"")); - - StepVerifier.create(session.receiveEvents()) - .expectErrorMatches( - error -> error instanceof IllegalStateException && error.getMessage().contains("Only one")) - .verify(); - session.close(); - assertFalse(session.isOpen()); } @Test @@ -626,48 +410,6 @@ public void syncConnectRejectsNullArguments() { assertEquals("'options' cannot be null.", optionsException.getMessage()); } - @Test - public void tokenFailureOccursBeforeNetworkAccess() { - AtomicBoolean connected = new AtomicBoolean(); - server = tlsServer().host("localhost").port(0).handle((request, response) -> { - connected.set(true); - return response.send(); - }).bindNow(); - TokenCredential credential = request -> Mono.error(new IllegalStateException("token unavailable")); - BetaVoiceAgentWebSocketAsyncClient client - = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") - .credential(credential) - .configuration(Configuration.NONE) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient(); - - StepVerifier.create(client.openWebSocketSession("agent", tlsOptions())) - .expectErrorMatches( - error -> error instanceof IllegalStateException && error.getMessage().contains("token unavailable")) - .verify(); - assertFalse(connected.get()); - } - - @Test - public void tokenAcquisitionDoesNotUseHandshakeTimeout() { - server = startServer(new CopyOnWriteArrayList<>(), new AtomicReference<>(), new AtomicReference<>(), - new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), false); - TokenCredential credential = request -> Mono.delay(Duration.ofMillis(1500)) - .map(ignored -> new AccessToken("test-token", OffsetDateTime.now().plusHours(1))); - BetaVoiceAgentWebSocketAsyncClient client - = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") - .credential(credential) - .configuration(Configuration.NONE) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient(); - VoiceAgentWebSocketConnectionOptions options = tlsOptions().setHandshakeTimeout(Duration.ofSeconds(1)); - - StepVerifier.withVirtualTime(() -> client.openWebSocketSession("agent", options).flatMap(session -> { - assertTrue(session.isOpen()); - return session.closeAsync(); - })).thenAwait(Duration.ofMillis(1500)).verifyComplete(); - } - @Test public void syncTokenFailureOccursBeforeNetworkAccess() { AtomicBoolean connected = new AtomicBoolean(); @@ -689,38 +431,6 @@ public void syncTokenFailureOccursBeforeNetworkAccess() { assertFalse(connected.get()); } - @Test - public void rejectedHandshakeMapsConflictToAzureException() { - server = tlsServer().host("localhost") - .port(0) - .handle( - (request, response) -> response.status(HttpResponseStatus.CONFLICT).sendString(Mono.just("conflict"))) - .bindNow(); - TokenCredential credential - = request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1))); - BetaVoiceAgentWebSocketAsyncClient client - = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") - .credential(credential) - .configuration(Configuration.NONE) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient(); - - StepVerifier.create(client.openWebSocketSession("disabled-agent", tlsOptions())).expectErrorSatisfies(error -> { - ResourceModifiedException exception = assertInstanceOf(ResourceModifiedException.class, error); - assertEquals(409, exception.getResponse().getStatusCode()); - }).verify(); - } - - @Test - public void nettyHandshakeResponseExposesBufferedBody() { - DefaultFullHttpResponse nettyResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, - HttpResponseStatus.CONFLICT, Unpooled.copiedBuffer("conflict", StandardCharsets.UTF_8)); - VoiceAgentWebSocketHttpResponse response - = new VoiceAgentWebSocketHttpResponse(URI.create("wss://example.com"), nettyResponse); - - assertEquals("conflict", response.getBodyAsString().block()); - } - @Test public void syncRejectedHandshakeMapsConflictToAzureException() { server = tlsServer().host("localhost") @@ -758,70 +468,6 @@ public void syncClientRejectsEmptyAgentName() { assertEquals("'agentName' cannot be empty.", exception.getMessage()); } - @Test - public void cancellingAsyncConnectCancelsTokenRequest() { - AtomicBoolean tokenRequestCancelled = new AtomicBoolean(); - TokenCredential credential - = request -> Mono.never().doOnCancel(() -> tokenRequestCancelled.set(true)); - BetaVoiceAgentWebSocketAsyncClient client = new AgentsClientBuilder().endpoint("https://example.com") - .credential(credential) - .configuration(Configuration.NONE) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient(); - - StepVerifier.create(client.openWebSocketSession("agent", tlsOptions())).thenCancel().verify(); - assertTrue(tokenRequestCancelled.get()); - } - - @Test - public void unknownEventFallsBackToRealtimeServerEvent() { - server = oneShotWebSocketServer("{\"type\":\"future.event\",\"value\":42}"); - BetaVoiceAgentWebSocketSessionAsyncClient session - = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); - - StepVerifier.create(session.receiveEvents()).assertNext(event -> { - assertEquals("future.event", event.getType().toString()); - RawRealtimeServerEvent raw = assertInstanceOf(RawRealtimeServerEvent.class, event); - assertEquals(42, raw.getRawEvent().toObject(Map.class).get("value")); - }).verifyComplete(); - session.close(); - } - - @Test - public void fragmentedTextFrameIsAggregated() { - String message = warningJson(); - int split = message.length() / 2; - Flux frames = Flux.just(new TextWebSocketFrame(false, 0, message.substring(0, split)), - new ContinuationWebSocketFrame(true, 0, message.substring(split))); - server = frameWebSocketServer(frames); - BetaVoiceAgentWebSocketSessionAsyncClient session - = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); - - StepVerifier.create(session.receiveEvents()).assertNext(this::assertWarningEvent).verifyComplete(); - session.close(); - } - - @Test - public void binaryJsonFrameIsParsed() { - server = frameWebSocketServer( - Mono.just(new BinaryWebSocketFrame(Unpooled.copiedBuffer(warningJson(), StandardCharsets.UTF_8)))); - BetaVoiceAgentWebSocketSessionAsyncClient session - = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); - - StepVerifier.create(session.receiveEvents()).assertNext(this::assertWarningEvent).verifyComplete(); - session.close(); - } - - @Test - public void malformedJsonTerminatesReceiveStream() { - server = oneShotWebSocketServer("{not-json"); - BetaVoiceAgentWebSocketSessionAsyncClient session - = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); - - StepVerifier.create(session.receiveEvents()).expectError().verify(); - assertFalse(session.isOpen()); - } - @Test public void syncReceiveBufferOverflowFailsTheEventStream() { Flux frames = Flux.range(0, 257).map(index -> new TextWebSocketFrame(warningJson())); @@ -873,49 +519,6 @@ public void syncOrderlyClosePreservesFullReceiveBuffer() { } } - @Test - public void closeIsIdempotentAndSendAfterCloseFails() { - List clientMessages = new CopyOnWriteArrayList<>(); - server = startServer(clientMessages, new AtomicReference<>(), new AtomicReference<>(), new AtomicReference<>(), - new AtomicReference<>(), new AtomicReference<>(), false); - BetaVoiceAgentWebSocketSessionAsyncClient session - = createAsyncClient(server.port()).openWebSocketSession("agent", tlsOptions()).block(); - - StepVerifier.create(session.closeAsync().then(session.closeAsync())).verifyComplete(); - StepVerifier.create(session.sendText("after close")) - .expectErrorMatches( - error -> error instanceof IllegalStateException && error.getMessage().contains("not open")) - .verify(); - } - - @Test - public void secureSessionUsesWssAndReceivesTypedEvent() throws Exception { - Http11SslContextSpec serverSsl = Http11SslContextSpec.forServer(TLS_CERTIFICATE.keyManagerFactory); - WebsocketServerSpec websocketSpec = WebsocketServerSpec.builder().protocols("realtime").build(); - server = tlsServer().host("localhost") - .port(0) - .secure(ssl -> ssl.sslContext(serverSsl)) - .handle((request, response) -> response.sendWebsocket( - (inbound, outbound) -> outbound.sendString(Mono.just(warningJson()), StandardCharsets.UTF_8) - .then(outbound.sendClose()), - websocketSpec)) - .bindNow(); - - TokenCredential credential - = request -> Mono.just(new AccessToken("tls-token", OffsetDateTime.now().plusHours(1))); - BetaVoiceAgentWebSocketSessionAsyncClient session - = new AgentsClientBuilder().endpoint("https://localhost:" + server.port() + "/api/projects/project") - .credential(credential) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient() - .openWebSocketSession("secure-agent", tlsOptions()) - .block(Duration.ofSeconds(5)); - - assertEquals("wss", session.getEndpoint().getScheme()); - StepVerifier.create(session.receiveEvents()).assertNext(this::assertWarningEvent).verifyComplete(); - session.close(); - } - @Test public void syncSessionReceivesTypedEventAndCloses() { List clientMessages = new CopyOnWriteArrayList<>(); @@ -954,16 +557,6 @@ public void syncSessionReceivesTypedEventAndCloses() { } } - private BetaVoiceAgentWebSocketAsyncClient createAsyncClient(int port) { - TokenCredential credential - = request -> Mono.just(new AccessToken("test-token", OffsetDateTime.now().plusHours(1))); - return new AgentsClientBuilder().endpoint("https://localhost:" + port + "/api/projects/project") - .credential(credential) - .configuration(Configuration.NONE) - .beta() - .buildBetaVoiceAgentWebSocketAsyncClient(); - } - private DisposableServer frameWebSocketServer(org.reactivestreams.Publisher frames) { WebsocketServerSpec spec = WebsocketServerSpec.builder().protocols("realtime").build(); return tlsServer().host("localhost") @@ -1034,6 +627,7 @@ private static TestCertificate create() { try { Path path = Files.createTempFile("voice-agent-websocket-", ".p12"); Files.delete(path); + path.toFile().deleteOnExit(); String password = UUID.randomUUID().toString(); String executable = Paths @@ -1076,13 +670,6 @@ private void installTrustStore() { } } - private void delete() { - try { - Files.deleteIfExists(path); - } catch (Exception error) { - throw new IllegalStateException(error); - } - } } private static SSLContext getDefaultSslContext() { @@ -1108,4 +695,5 @@ private static String decode(String value) { throw new IllegalStateException(error); } } + }