Negotiate HTTP/2 from the first request over plain http

The Streamable HTTP transport requires HTTP/2, and localhost without TLS
is a first-class deployment. Over cleartext the JDK HttpClient only
offers the h2c upgrade on a request without a body, so initialize, a
POST, went out on HTTP/1.1 and the connection upgraded only on the first
SSE GET. The client now sends a bodiless OPTIONS before initialize on
http:// endpoints; a failed probe is ignored. A new test drives the SDK
client against the SDK server over http:// through a recording
HttpClient and asserts every response, initialize, POSTs and the SSE
streams, was HTTP/2; it fails without the probe.

Author: Mark Pollack

CHANGELOG.md

@@ -63,6 +63,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   the SSE mailbox and per-subscriber queue limits are configurable, attached streams get a `: keep-alive`
   comment every 15 s so proxies do not cut idle connections, and a new GET on a stream takes it over
   from a subscriber the server may not yet know is dead instead of fanning out duplicates.
+- **HTTP/2 over plain `http://`.** The RFD requires HTTP/2, and localhost without TLS is a first-class
+  deployment. Over cleartext the JDK client only offers the h2c upgrade on a request without a body, so
+  `initialize`, a POST, went out on HTTP/1.1. `StreamableHttpAcpClientTransport` now sends a bodiless
+  OPTIONS first on `http://` endpoints, and every request, streams included, runs on HTTP/2.
 - **Client sessions learn that their transport died.** `AcpClientTransport.awaitTermination()` (default:
   never) is implemented by the Streamable HTTP and WebSocket client transports; `AcpClientSession` fails
   pending requests at once with the cause, and every later request, instead of waiting out the request

acp-core/src/main/java/com/agentclientprotocol/sdk/client/transport/StreamableHttpAcpClientTransport.java

@@ -298,6 +298,30 @@ public Mono<Void> sendMessage(JSONRPCMessage message) {
 		return routeAndPost(message);
 	}
 
+	/**
+	 * The transport requires HTTP/2 (RFD). Over {@code https} ALPN negotiates it on the first
+	 * request. Over cleartext {@code http} the JDK client only offers the h2c upgrade on a
+	 * request without a body, and {@code initialize} is a POST, so without this the whole
+	 * connection would stay on HTTP/1.1. A bodiless OPTIONS first upgrades the connection;
+	 * every later request, the POSTs and the SSE GETs, reuses it over HTTP/2. A failed probe
+	 * is ignored: initialize then proceeds and reports any real problem itself.
+	 */
+	private Mono<Void> upgradeCleartextToHttp2() {
+		if (!"http".equalsIgnoreCase(endpointUri.getScheme()) || httpClient.version() != HttpClient.Version.HTTP_2) {
+			return Mono.empty();
+		}
+		HttpRequest probe = HttpRequest.newBuilder(endpointUri)
+			.method("OPTIONS", HttpRequest.BodyPublishers.noBody())
+			.build();
+		return sendAsync(probe, HttpResponse.BodyHandlers.discarding())
+			.doOnNext(response -> logger.debug("Cleartext probe to {} negotiated {}", endpointUri, response.version()))
+			.then()
+			.onErrorResume(error -> {
+				logger.debug("Cleartext HTTP/2 probe to {} failed: {}", endpointUri, error.getMessage());
+				return Mono.empty();
+			});
+	}
+
 	private Mono<Void> initialize(AcpSchema.JSONRPCRequest request) {
 		if (!initialized.compareAndSet(false, true)) {
 			return Mono.error(new IllegalStateException("Transport is already initialized"));
@@ -314,7 +338,8 @@ private Mono<Void> initialize(AcpSchema.JSONRPCRequest request) {
 			return Mono.error(new AcpConnectionException("Failed to serialize initialize request", e));
 		}
 
-		return sendAsync(httpRequest, HttpResponse.BodyHandlers.ofString())
+		return upgradeCleartextToHttp2()
+			.then(sendAsync(httpRequest, HttpResponse.BodyHandlers.ofString()))
 			.flatMap(response -> {
 				if (response.statusCode() != 200) {
 					return Mono.error(new AcpConnectionException(

acp-streamable-http-jetty/src/test/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpClientServerIntegrationTest.java

@@ -210,6 +210,132 @@ void initializeIsServedOverHttp2() throws Exception {
 		}
 	}
 
+	/**
+	 * Localhost over plain {@code http://} is a first-class deployment (no TLS), and the RFD
+	 * requires HTTP/2: every request the SDK client makes, initialize, the POSTs and the
+	 * long-lived SSE GETs, must run on HTTP/2 over cleartext.
+	 */
+	@Test
+	void sdkClientUsesHttp2ForEveryRequestOverCleartextHttp() throws Exception {
+		StreamableHttpAcpAgentTransport server = startServer(StreamableHttpAcpAgentTransportOptions.defaults());
+		RecordingHttpClient http = new RecordingHttpClient(
+				HttpClient.newBuilder().version(HttpClient.Version.HTTP_2).build());
+		AcpAsyncClient client = AcpClient
+			.async(new StreamableHttpAcpClientTransport(endpoint(server), AcpJsonMapper.createDefault(), http))
+			.requestTimeout(TIMEOUT)
+			.build();
+		try {
+			client.initialize().block(TIMEOUT);
+			AcpSchema.NewSessionResponse session = client
+				.newSession(new AcpSchema.NewSessionRequest("/workspace", List.of()))
+				.block(TIMEOUT);
+			client.prompt(new AcpSchema.PromptRequest(session.sessionId(), List.of(new AcpSchema.TextContent("hi"))))
+				.block(TIMEOUT);
+
+			assertThat(endpoint(server).getScheme()).isEqualTo("http");
+			assertThat(http.versions()).as("request methods seen: " + http.methods()).isNotEmpty();
+			assertThat(http.methods()).contains("POST", "GET");
+			assertThat(http.versions()).as("every response, including SSE streams").containsOnly(HttpClient.Version.HTTP_2);
+		}
+		finally {
+			client.closeGracefully().block(TIMEOUT);
+			server.closeGracefully().block(TIMEOUT);
+		}
+	}
+
+	/** Delegates to a real client and records the method and negotiated version of every response. */
+	private static final class RecordingHttpClient extends HttpClient {
+
+		private final HttpClient delegate;
+
+		private final List<HttpClient.Version> versions = new CopyOnWriteArrayList<>();
+
+		private final List<String> methods = new CopyOnWriteArrayList<>();
+
+		RecordingHttpClient(HttpClient delegate) {
+			this.delegate = delegate;
+		}
+
+		List<HttpClient.Version> versions() {
+			return versions;
+		}
+
+		List<String> methods() {
+			return methods;
+		}
+
+		private <T> HttpResponse<T> record(HttpRequest request, HttpResponse<T> response) {
+			methods.add(request.method());
+			versions.add(response.version());
+			return response;
+		}
+
+		@Override
+		public java.util.Optional<java.net.CookieHandler> cookieHandler() {
+			return delegate.cookieHandler();
+		}
+
+		@Override
+		public java.util.Optional<Duration> connectTimeout() {
+			return delegate.connectTimeout();
+		}
+
+		@Override
+		public Redirect followRedirects() {
+			return delegate.followRedirects();
+		}
+
+		@Override
+		public java.util.Optional<java.net.ProxySelector> proxy() {
+			return delegate.proxy();
+		}
+
+		@Override
+		public javax.net.ssl.SSLContext sslContext() {
+			return delegate.sslContext();
+		}
+
+		@Override
+		public javax.net.ssl.SSLParameters sslParameters() {
+			return delegate.sslParameters();
+		}
+
+		@Override
+		public java.util.Optional<java.net.Authenticator> authenticator() {
+			return delegate.authenticator();
+		}
+
+		@Override
+		public Version version() {
+			return delegate.version();
+		}
+
+		@Override
+		public java.util.Optional<java.util.concurrent.Executor> executor() {
+			return delegate.executor();
+		}
+
+		@Override
+		public <T> HttpResponse<T> send(HttpRequest request, HttpResponse.BodyHandler<T> handler)
+				throws IOException, InterruptedException {
+			return record(request, delegate.send(request, handler));
+		}
+
+		@Override
+		public <T> java.util.concurrent.CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request,
+				HttpResponse.BodyHandler<T> handler) {
+			return delegate.sendAsync(request, handler).thenApply(response -> record(request, response));
+		}
+
+		@Override
+		public <T> java.util.concurrent.CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request,
+				HttpResponse.BodyHandler<T> handler, HttpResponse.PushPromiseHandler<T> pushPromiseHandler) {
+			return delegate.sendAsync(request, handler, pushPromiseHandler)
+				.thenApply(response -> record(request, response));
+		}
+
+	}
+
 	/** An agent factory that fails must fail the WebSocket upgrade, not leave a half-open connection. */
 	@Test
 	void webSocketUpgradeFailsCleanlyWhenTheAgentFactoryThrows() throws Exception {