Interoperate with the Rust, TypeScript and Python HTTP transports

An interop run over plain http against the three other SDKs found four
Java-side problems:

- Against a cleartext server without h2c, the client kept offering
  Upgrade: h2c on every request; the TypeScript example server routes
  any request with an Upgrade header to its WebSocket handler and
  answered initialize with 405. After the OPTIONS probe settles on
  HTTP/1.1 (or fails, or takes over five seconds) every request is now
  pinned to HTTP/1.1, including the initialize request built before it.
- session/load always opened a new GET for the session stream, even when
  it was open; the TypeScript and Rust servers allow one receiver per
  stream and answered 409. An open stream is now reused.
- The Rust and Python clients post every response without
  Acp-Session-Id, and the Python client sends session/load without it
  (the RFD is ambiguous there). The server accepts a missing header in
  both cases, taking the session from params; a header naming a
  different session is still rejected.
- The provisional-session limit is checked and reserved atomically.

The initialize response's Acp-Connection-Id requirement regains the test
it lost with the fixture server.

Author: Mark Pollack

CHANGELOG.md

@@ -33,7 +33,9 @@ building a second client on an already-connected transport now fails at construc
 - **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.
+  OPTIONS first on `http://` endpoints, so against a server that speaks h2c (the SDK's own does) every
+  request, streams included, runs on HTTP/2. Against one that does not, the probe settles on HTTP/1.1
+  and later requests stop offering the upgrade, which some servers route to their WebSocket handler.
 - **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

@@ -40,6 +40,7 @@
 import com.agentclientprotocol.sdk.spec.AcpClientTransport;
 import com.agentclientprotocol.sdk.spec.AcpSchema;
 import com.agentclientprotocol.sdk.spec.AcpSchema.JSONRPCMessage;
+import com.agentclientprotocol.sdk.util.AcpSchedulers;
 import com.agentclientprotocol.sdk.util.Assert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -299,12 +300,23 @@ public Mono<Void> sendMessage(JSONRPCMessage 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.
+	 * HTTP version pinned for every request after the cleartext probe, or {@code null} to
+	 * use the client's own setting. Set to HTTP/1.1 when an {@code http://} server does not
+	 * speak h2c, so later requests stop carrying {@code Upgrade: h2c}: some servers hand any
+	 * request with an Upgrade header to their WebSocket handler and answer 405 (the
+	 * TypeScript SDK's example server does).
+	 */
+	private volatile HttpClient.Version pinnedVersion;
+
+	private static final Duration PROBE_TIMEOUT = Duration.ofSeconds(5);
+
+	/**
+	 * The transport requires HTTP/2 (RFD). Over {@code https} ALPN negotiates it. Over
+	 * cleartext {@code http} the JDK offers an h2c upgrade on every request, but servers
+	 * (Jetty among them) only honour it on a request without a body, and {@code initialize}
+	 * is a POST. A bodiless OPTIONS first upgrades the connection when the server speaks
+	 * h2c; every later request reuses it over HTTP/2. When it does not (answer on HTTP/1.1,
+	 * an error, or no answer within five seconds), every later request is pinned to HTTP/1.1.
 	 */
 	private Mono<Void> upgradeCleartextToHttp2() {
 		if (!"http".equalsIgnoreCase(endpointUri.getScheme()) || httpClient.version() != HttpClient.Version.HTTP_2) {
@@ -313,15 +325,41 @@ private Mono<Void> upgradeCleartextToHttp2() {
 		HttpRequest probe = HttpRequest.newBuilder(endpointUri)
 			.method("OPTIONS", HttpRequest.BodyPublishers.noBody())
 			.build();
+		// Cancelling the Mono on timeout cancels the HTTP exchange (see sendAsync).
 		return sendAsync(probe, HttpResponse.BodyHandlers.discarding())
-			.doOnNext(response -> logger.debug("Cleartext probe to {} negotiated {}", endpointUri, response.version()))
+			.timeout(PROBE_TIMEOUT, AcpSchedulers.timeouts())
+			.doOnNext(response -> {
+				logger.debug("Cleartext probe to {} negotiated {}", endpointUri, response.version());
+				if (response.version() != HttpClient.Version.HTTP_2) {
+					this.pinnedVersion = HttpClient.Version.HTTP_1_1;
+				}
+			})
 			.then()
 			.onErrorResume(error -> {
-				logger.debug("Cleartext HTTP/2 probe to {} failed: {}", endpointUri, error.getMessage());
+				logger.debug("Cleartext HTTP/2 probe to {} failed ({}); using HTTP/1.1", endpointUri, error.getMessage());
+				this.pinnedVersion = HttpClient.Version.HTTP_1_1;
 				return Mono.empty();
 			});
 	}
 
+	/** Re-stamps a request built before the probe with the version the probe settled on. */
+	private HttpRequest pinned(HttpRequest request) {
+		HttpClient.Version version = this.pinnedVersion;
+		if (version == null) {
+			return request;
+		}
+		return HttpRequest.newBuilder(request, (name, value) -> true).version(version).build();
+	}
+
+	private HttpRequest.Builder newRequest() {
+		HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(endpointUri);
+		HttpClient.Version version = this.pinnedVersion;
+		if (version != null) {
+			requestBuilder.version(version);
+		}
+		return requestBuilder;
+	}
+
 	private Mono<Void> initialize(AcpSchema.JSONRPCRequest request) {
 		if (!initialized.compareAndSet(false, true)) {
 			return Mono.error(new IllegalStateException("Transport is already initialized"));
@@ -338,8 +376,9 @@ private Mono<Void> initialize(AcpSchema.JSONRPCRequest request) {
 			return Mono.error(new AcpConnectionException("Failed to serialize initialize request", e));
 		}
 
+		HttpRequest initializeRequest = httpRequest;
 		return upgradeCleartextToHttp2()
-			.then(sendAsync(httpRequest, HttpResponse.BodyHandlers.ofString()))
+			.then(Mono.defer(() -> sendAsync(pinned(initializeRequest), HttpResponse.BodyHandlers.ofString())))
 			.flatMap(response -> {
 				if (response.statusCode() != 200) {
 					return Mono.error(new AcpConnectionException(
@@ -397,6 +436,13 @@ private Mono<Void> routeAndPost(JSONRPCMessage message) {
 	private Mono<Void> prepareRoute(ResolvedOutboundRoute resolved) {
 		if (resolved.message() instanceof AcpSchema.JSONRPCRequest request
 				&& AcpSchema.METHOD_SESSION_LOAD.equals(request.method())) {
+			// Open the session stream first (the RFD's reconnect order) unless it is already
+			// open: servers allow one receiver per stream, and the TypeScript and Rust servers
+			// answer a second GET with 409.
+			SseStream existing = sessionStreams.get(resolved.scope().sessionId());
+			if (existing != null && !existing.closed.get()) {
+				return Mono.empty();
+			}
 			return openSessionStream(resolved.scope().sessionId());
 		}
 		if (resolved.scope().isSession() && !sessionStreams.containsKey(resolved.scope().sessionId())) {
@@ -427,7 +473,7 @@ private Mono<Void> postAccepted(JSONRPCMessage message, RouteScope scope) {
 	}
 
 	private HttpRequest.Builder jsonPostBuilder(RouteScope scope) {
-		HttpRequest.Builder builder = HttpRequest.newBuilder(endpointUri)
+		HttpRequest.Builder builder = newRequest()
 			.header("Content-Type", CONTENT_TYPE_JSON)
 			.header("Accept", CONTENT_TYPE_JSON);
 		addScopeHeaders(builder, scope);
@@ -473,7 +519,7 @@ private Mono<Void> createSessionStreamOpenMono(String sessionId) {
 	}
 
 	private Mono<SseStream> openSseStream(RouteScope scope) {
-		HttpRequest.Builder builder = HttpRequest.newBuilder(endpointUri).GET().header("Accept", CONTENT_TYPE_EVENT_STREAM);
+		HttpRequest.Builder builder = newRequest().GET().header("Accept", CONTENT_TYPE_EVENT_STREAM);
 		addScopeHeaders(builder, scope);
 		HttpRequest request = builder.build();
 
@@ -687,7 +733,7 @@ public Mono<Void> closeGracefully() {
 
 			Mono<Void> deleteRequest = Mono.empty();
 			if (connectionId != null) {
-				HttpRequest request = HttpRequest.newBuilder(endpointUri)
+				HttpRequest request = newRequest()
 					.DELETE()
 					.header(HEADER_CONNECTION_ID, connectionId)
 					.build();

acp-core/src/test/java/com/agentclientprotocol/sdk/client/transport/StreamableHttpAcpClientTransportTest.java

@@ -633,6 +633,11 @@ void reopensClosedSessionSseBeforePostingNextSessionRequest() throws Exception {
 	 */
 	private HttpClient twoStreamHttpClient(PipedInputStream connectionStreamBody,
 			Map<String, PipedOutputStream> sessionWriters) {
+		return twoStreamHttpClient(connectionStreamBody, sessionWriters, new java.util.concurrent.CopyOnWriteArrayList<>());
+	}
+
+	private HttpClient twoStreamHttpClient(PipedInputStream connectionStreamBody,
+			Map<String, PipedOutputStream> sessionWriters, List<String> sessionGets) {
 		HttpClient httpClient = mock(HttpClient.class);
 		when(httpClient.sendAsync(any(), any())).thenAnswer(invocation -> {
 			HttpRequest request = invocation.getArgument(0);
@@ -648,6 +653,7 @@ private HttpClient twoStreamHttpClient(PipedInputStream connectionStreamBody,
 			}
 			if ("GET".equals(request.method())) {
 				String sessionId = request.headers().firstValue("Acp-Session-Id").orElseThrow();
+				sessionGets.add(sessionId);
 				PipedInputStream sessionBody = new PipedInputStream(16 * 1024);
 				sessionWriters.put(sessionId, new PipedOutputStream(sessionBody));
 				return CompletableFuture.completedFuture(
@@ -739,6 +745,118 @@ void promptResponseArrivingOnTheConnectionStreamIsDelivered() throws Exception {
 		}
 	}
 
+	/**
+	 * Servers allow one receiver per stream; the TypeScript and Rust servers answer a second
+	 * GET with 409. session/load for a session whose stream is open must not open another.
+	 */
+	@Test
+	void loadingASessionWhoseStreamIsOpenDoesNotOpenItAgain() throws Exception {
+		Map<String, PipedOutputStream> sessionWriters = new ConcurrentHashMap<>();
+		List<String> sessionGets = new java.util.concurrent.CopyOnWriteArrayList<>();
+		PipedInputStream connectionStreamBody = new PipedInputStream();
+		PipedOutputStream connectionStreamWriter = new PipedOutputStream(connectionStreamBody);
+		BlockingQueue<AcpSchema.JSONRPCMessage> inboundMessages = new LinkedBlockingQueue<>();
+		StreamableHttpAcpClientTransport transport = new StreamableHttpAcpClientTransport(
+				URI.create("https://localhost:8443/acp"), jsonMapper,
+				twoStreamHttpClient(connectionStreamBody, sessionWriters, sessionGets));
+		try {
+			transport.connect(message -> message.doOnNext(inboundMessages::add).then(Mono.empty())).block();
+			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_INITIALIZE, "init-1",
+					AcpTestFixtures.createInitializeRequest())).block();
+			awaitResponse(inboundMessages, "init-1");
+			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_SESSION_NEW, "new-1",
+					AcpTestFixtures.createNewSessionRequest("/workspace"))).block();
+			writeSse(connectionStreamWriter, new AcpSchema.JSONRPCResponse(AcpSchema.JSONRPC_VERSION, "new-1",
+					new AcpSchema.NewSessionResponse("sess-1", null, null), null));
+			awaitResponse(inboundMessages, "new-1");
+			assertThat(sessionGets).containsExactly("sess-1");
+
+			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_SESSION_LOAD, "load-1",
+					new AcpSchema.LoadSessionRequest("sess-1", "/workspace", List.of()))).block();
+			writeSse(connectionStreamWriter, new AcpSchema.JSONRPCResponse(AcpSchema.JSONRPC_VERSION, "load-1",
+					new AcpSchema.LoadSessionResponse(null, null), null));
+			awaitResponse(inboundMessages, "load-1");
+
+			assertThat(sessionGets).as("no second GET for an open session stream").containsExactly("sess-1");
+		}
+		finally {
+			connectionStreamWriter.close();
+			sessionWriters.values().forEach(writer -> {
+				try {
+					writer.close();
+				}
+				catch (Exception ignored) {
+				}
+			});
+			transport.close();
+		}
+	}
+
+	/**
+	 * Over http:// a server that does not speak h2c must not keep receiving Upgrade headers:
+	 * the TypeScript example server routes any request carrying one to its WebSocket handler
+	 * and answers 405. After a probe that comes back on HTTP/1.1, every request is pinned.
+	 */
+	@Test
+	void cleartextServerWithoutH2cGetsPlainHttp11Requests() throws Exception {
+		HttpClient httpClient = mock(HttpClient.class);
+		when(httpClient.version()).thenReturn(HttpClient.Version.HTTP_2);
+		List<HttpRequest> requests = new java.util.concurrent.CopyOnWriteArrayList<>();
+		when(httpClient.sendAsync(any(), any())).thenAnswer(invocation -> {
+			HttpRequest request = invocation.getArgument(0);
+			requests.add(request);
+			if ("OPTIONS".equals(request.method())) {
+				HttpResponse<Object> probe = response(405, Map.of(), null);
+				when(probe.version()).thenReturn(HttpClient.Version.HTTP_1_1);
+				return CompletableFuture.completedFuture(probe);
+			}
+			String initializeResponse = jsonMapper.writeValueAsString(AcpTestFixtures
+				.createJsonRpcResponse("init-1", AcpTestFixtures.createInitializeResponse()));
+			if ("POST".equals(request.method())) {
+				return CompletableFuture.completedFuture(response(200,
+						Map.of("Content-Type", "application/json", "Acp-Connection-Id", "conn-1"), initializeResponse));
+			}
+			return CompletableFuture.completedFuture(
+					response(200, Map.of("Content-Type", "text/event-stream"), emptyBody()));
+		});
+		StreamableHttpAcpClientTransport transport = new StreamableHttpAcpClientTransport(
+				URI.create("http://localhost:8080/acp"), jsonMapper, httpClient);
+		transport.setExceptionHandler(error -> {
+		});
+		try {
+			transport.connect(message -> message.then(Mono.empty())).block();
+			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_INITIALIZE, "init-1",
+					AcpTestFixtures.createInitializeRequest())).block();
+
+			assertThat(requests.get(0).method()).isEqualTo("OPTIONS");
+			assertThat(requests.subList(1, requests.size()))
+				.as("every request after the probe is pinned to HTTP/1.1")
+				.isNotEmpty()
+				.allMatch(request -> request.version().equals(java.util.Optional.of(HttpClient.Version.HTTP_1_1)));
+		}
+		finally {
+			transport.close();
+		}
+	}
+
+	@Test
+	void initializeRequiresConnectionIdHeader() throws Exception {
+		HttpClient httpClient = mock(HttpClient.class);
+		String body = jsonMapper.writeValueAsString(
+				AcpTestFixtures.createJsonRpcResponse("init-1", AcpTestFixtures.createInitializeResponse()));
+		HttpResponse<Object> noConnectionId = response(200, Map.of("Content-Type", "application/json"), body);
+		when(httpClient.sendAsync(any(), any())).thenReturn(CompletableFuture.completedFuture(noConnectionId));
+		StreamableHttpAcpClientTransport transport = new StreamableHttpAcpClientTransport(
+				URI.create("https://localhost:8443/acp"), jsonMapper, httpClient);
+		transport.setExceptionHandler(error -> {
+		});
+
+		assertThatThrownBy(() -> transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_INITIALIZE,
+				"init-1", AcpTestFixtures.createInitializeRequest())).block())
+			.isInstanceOf(AcpConnectionException.class)
+			.hasMessageContaining("Acp-Connection-Id");
+	}
+
 	private void awaitSessionSseClosure() throws InterruptedException {
 		Thread.sleep(100);
 	}

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpConnection.java

@@ -249,7 +249,16 @@ private void validateClientResponseScope(AcpSchema.JSONRPCResponse response, Str
 			logger.warn("Client posted response for unknown agent request id {}", response.id());
 			return;
 		}
-		RouteScope actual = sessionHeader == null ? RouteScope.connection() : RouteScope.session(sessionHeader);
+		if (sessionHeader == null) {
+			// The RFD asks for Acp-Session-Id on permission responses; the Rust and Python
+			// clients omit it on every response. The id alone identifies the exchange, so a
+			// missing header is accepted. A header naming a different scope is still an error.
+			logger.debug("Client response {} carried no {}; accepting it for {}", response.id(),
+					StreamableHttpAcpAgentTransport.HEADER_SESSION_ID, expected);
+			agentRequestRoutes.remove(response.id(), expected);
+			return;
+		}
+		RouteScope actual = RouteScope.session(sessionHeader);
 		if (!Objects.equals(expected, actual)) {
 			throw new AcpConnectionException(
 					"Response id " + response.id() + " arrived on " + actual + " but expected " + expected);
@@ -272,7 +281,7 @@ private SseOutboundStream openSessionStream(String sessionId) {
 	}
 
 	/** Provisional sessions are bounded: a client cannot grow state with arbitrary ids. */
-	private void addProvisionalSession(String sessionId) {
+	private synchronized void addProvisionalSession(String sessionId) {
 		long provisional = sessions.values().stream().filter(state -> state == SessionState.PENDING_LOAD).count();
 		if (provisional >= options.maxProvisionalSessions()
 				&& sessions.get(sessionId) != SessionState.PENDING_LOAD) {

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpRouting.java

@@ -141,7 +141,10 @@ else if (message instanceof AcpSchema.JSONRPCNotification notification) {
 				break;
 			case AcpSchema.METHOD_SESSION_LOAD:
 			case AcpSchema.METHOD_SESSION_RESUME:
-				requestScope = requireSessionScope(method, params, sessionHeader);
+				// The RFD is ambiguous here: its reconnect diagram sends the header, its text
+				// says load answers on the connection stream because the client has no session
+				// yet. The Python client omits it. Accept either; the id comes from params.
+				requestScope = sessionScopeFromParams(method, params, sessionHeader);
 				kind = RequestKind.SESSION_LOAD;
 				responseScope = RouteScope.connection();
 				break;
@@ -174,6 +177,14 @@ else if (message instanceof AcpSchema.JSONRPCNotification notification) {
 		return new ResolvedInboundRoute(message, requestScope, requestRoute);
 	}
 
+	/** Like {@link #requireSessionScope} but tolerates a missing header; a conflicting one is still rejected. */
+	RouteScope sessionScopeFromParams(String method, Object params, String sessionHeader) {
+		if (sessionHeader == null) {
+			return RouteScope.session(requireSessionId(params, method));
+		}
+		return requireSessionScope(method, params, sessionHeader);
+	}
+
 	RouteScope requireSessionScope(String method, Object params, String sessionHeader) {
 		String sessionId = requireSessionId(params, method);
 		if (sessionHeader == null) {