Skip to content

Commit 4d0d012

Browse files
kamikaz1kkamikaz1k
committed
refactor: remove streamable HTTP strict routing
1 parent 8714d91 commit 4d0d012

5 files changed

Lines changed: 15 additions & 169 deletions

File tree

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

Lines changed: 0 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -68,25 +68,6 @@ public class StreamableHttpAcpClientTransport implements AcpClientTransport {
6868

6969
private static final Duration CLOSE_TIMEOUT = Duration.ofSeconds(10);
7070

71-
/**
72-
* Controls how unknown outbound request / notification methods are classified.
73-
*/
74-
public enum RoutingMode {
75-
76-
/**
77-
* Prefer explicit ACP routing, but fall back to session-id shape inference for
78-
* unknown methods so clients can remain forward-compatible with extensions.
79-
*/
80-
COMPATIBLE,
81-
82-
/**
83-
* Require every outbound request / notification method to have an explicit routing
84-
* rule.
85-
*/
86-
STRICT
87-
88-
}
89-
9071
private enum ScopeKind {
9172

9273
BOOTSTRAP,
@@ -177,8 +158,6 @@ private record HttpClientBundle(HttpClient httpClient, ExecutorService ownedExec
177158

178159
private volatile String connectionId;
179160

180-
private volatile RoutingMode routingMode = RoutingMode.COMPATIBLE;
181-
182161
private volatile Consumer<Throwable> exceptionHandler = t -> logger.error("Transport error", t);
183162

184163
/**
@@ -243,17 +222,6 @@ private static HttpClientBundle createDefaultHttpClient() {
243222
return new HttpClientBundle(client, executor);
244223
}
245224

246-
/**
247-
* Sets the routing mode for outbound request / notification classification.
248-
* @param routingMode routing mode to apply
249-
* @return this transport
250-
*/
251-
public StreamableHttpAcpClientTransport routingMode(RoutingMode routingMode) {
252-
Assert.notNull(routingMode, "The routingMode can not be null");
253-
this.routingMode = routingMode;
254-
return this;
255-
}
256-
257225
@Override
258226
public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
259227
Assert.notNull(handler, "The handler can not be null");
@@ -268,25 +236,9 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
268236
private void handleIncomingMessages(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
269237
this.inboundSink.asFlux()
270238
.flatMap(message -> Mono.just(message).transform(handler))
271-
.doOnNext(this::forwardHandlerEmissionForCompatibility)
272239
.subscribe();
273240
}
274241

275-
private void forwardHandlerEmissionForCompatibility(JSONRPCMessage emittedMessage) {
276-
/*
277-
* Compatibility note:
278-
* WebSocketAcpClientTransport currently forwards any message emitted by the
279-
* registered client handler back onto the transport. AcpClientSession also sends
280-
* client responses explicitly via sendMessage(...), so the client-side contract is
281-
* still ambiguous. Preserve parity for now and keep this path isolated so it can be
282-
* removed cheaply if the client transport contract is later made receive-only.
283-
*/
284-
if (emittedMessage != null && !closing.get()) {
285-
routeAndPost(emittedMessage).subscribe(v -> {
286-
}, exceptionHandler);
287-
}
288-
}
289-
290242
@Override
291243
public Mono<Void> sendMessage(JSONRPCMessage message) {
292244
Assert.notNull(message, "The message can not be null");
@@ -518,9 +470,6 @@ private ResolvedOutboundRoute resolveRequestOrNotificationRoute(JSONRPCMessage m
518470
break;
519471
default:
520472
Optional<String> sessionId = extractSessionId(params);
521-
if (routingMode == RoutingMode.STRICT) {
522-
throw new AcpConnectionException("No explicit routing rule for outbound method " + method);
523-
}
524473
if (sessionId.isPresent()) {
525474
logger.warn("Falling back to inferred session routing for unknown method '{}'", method);
526475
requestScope = RouteScope.session(sessionId.get());

‎acp-core/src/main/java/com/agentclientprotocol/sdk/spec/AcpClientSession.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -148,15 +148,6 @@ public AcpClientSession(Duration requestTimeout, AcpClientTransport transport,
148148
return t;
149149
}), "acp-timeout-" + sessionPrefix);
150150

151-
/*
152-
* Client transports currently retain a compatibility path that may forward any
153-
* message emitted by this handler back onto the wire. The session handles outbound
154-
* replies explicitly via transport.sendMessage(...), so the default session handler
155-
* should consume inbound messages without re-emitting them. The transport-level
156-
* handler type is Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>>, so returning
157-
* Mono.empty() is intentional here: the signature permits an emitted message, but
158-
* the default client session has no message to return through that path.
159-
*/
160151
this.transport.connect(mono -> mono.doOnNext(this::handle).then(Mono.empty())).transform(connectHook).subscribe();
161152
}
162153

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

Lines changed: 0 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -85,35 +85,11 @@ void constructorAcceptsCustomHttpClient() {
8585
assertThat(transport).isNotNull();
8686
}
8787

88-
@Test
89-
void routingModeIsConfigurable() {
90-
StreamableHttpAcpClientTransport transport = new StreamableHttpAcpClientTransport(
91-
URI.create("https://localhost:8443/acp"), jsonMapper)
92-
.routingMode(StreamableHttpAcpClientTransport.RoutingMode.STRICT);
93-
94-
assertThat(transport).isNotNull();
95-
}
96-
9788
@Test
9889
void defaultAcpPathIsCorrect() {
9990
assertThat(StreamableHttpAcpClientTransport.DEFAULT_ACP_PATH).isEqualTo("/acp");
10091
}
10192

102-
@Test
103-
void strictRoutingRejectsUnknownOutboundMethods() {
104-
StreamableHttpAcpClientTransport transport = new StreamableHttpAcpClientTransport(
105-
URI.create("https://localhost:8443/acp"), jsonMapper)
106-
.routingMode(StreamableHttpAcpClientTransport.RoutingMode.STRICT);
107-
108-
transport.connect(message -> Mono.empty()).block();
109-
110-
assertThatThrownBy(() -> transport
111-
.sendMessage(new AcpSchema.JSONRPCNotification(AcpSchema.JSONRPC_VERSION, "extension/custom",
112-
Map.of("sessionId", "session-1")))
113-
.block())
114-
.hasMessageContaining("No explicit routing rule for outbound method extension/custom");
115-
}
116-
11793
@Test
11894
void concurrentSessionLoadsReuseInFlightSessionStreamOpen() throws Exception {
11995
HttpClient httpClient = mock(HttpClient.class);

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

Lines changed: 1 addition & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -93,25 +93,6 @@ public class StreamableHttpAcpAgentTransport {
9393

9494
private static final Duration INITIALIZE_TIMEOUT = Duration.ofSeconds(30);
9595

96-
/**
97-
* Controls whether unknown message methods may fall back to shape-based routing.
98-
*/
99-
public enum RoutingMode {
100-
101-
/**
102-
* Prefer explicit ACP routing and fall back to session-id shape inference for
103-
* extension methods. Also permits provisional session streams before
104-
* {@code session/load} so the currently ambiguous resume flow can work.
105-
*/
106-
COMPATIBLE,
107-
108-
/**
109-
* Require explicit routing rules and reject unknown session streams.
110-
*/
111-
STRICT
112-
113-
}
114-
11596
private enum ScopeKind {
11697

11798
CONNECTION,
@@ -181,8 +162,6 @@ private record ResolvedInboundRoute(JSONRPCMessage message, RouteScope requestSc
181162

182163
private final Sinks.One<Void> terminationSink = Sinks.one();
183164

184-
private volatile RoutingMode routingMode = RoutingMode.COMPATIBLE;
185-
186165
private volatile Server server;
187166

188167
private volatile ServerConnector connector;
@@ -216,17 +195,6 @@ public StreamableHttpAcpAgentTransport(int port, String path, AcpJsonMapper json
216195
this.agentFactory = agentFactory;
217196
}
218197

219-
/**
220-
* Sets the routing mode used by the listener.
221-
* @param routingMode routing mode to use
222-
* @return this transport
223-
*/
224-
public StreamableHttpAcpAgentTransport routingMode(RoutingMode routingMode) {
225-
Assert.notNull(routingMode, "The routingMode can not be null");
226-
this.routingMode = routingMode;
227-
return this;
228-
}
229-
230198
/**
231199
* Starts the embedded Jetty server.
232200
* @return a mono that completes when the listener is ready
@@ -667,9 +635,6 @@ private RouteScope resolveAgentRequestOrNotificationScope(String method, Object
667635
return RouteScope.session(requireSessionId(params, method));
668636
default:
669637
Optional<String> sessionId = extractSessionId(params);
670-
if (routingMode == RoutingMode.STRICT) {
671-
throw new AcpConnectionException("No explicit routing rule for outbound method " + method);
672-
}
673638
return sessionId.map(RouteScope::session).orElseGet(RouteScope::connection);
674639
}
675640
}
@@ -714,9 +679,6 @@ else if (message instanceof AcpSchema.JSONRPCNotification notification) {
714679
break;
715680
default:
716681
Optional<String> sessionId = extractSessionId(params);
717-
if (routingMode == RoutingMode.STRICT) {
718-
throw new AcpConnectionException("No explicit routing rule for inbound method " + method);
719-
}
720682
if (sessionId.isPresent()) {
721683
requestScope = requireSessionScope(method, params, sessionHeader);
722684
}
@@ -746,9 +708,6 @@ private void prepareSessionForInbound(String sessionId, ClientRequestRoute route
746708
SessionState current = sessions.get(sessionId);
747709
if (route != null && route.kind() == RequestKind.SESSION_LOAD) {
748710
if (current == null) {
749-
if (routingMode == RoutingMode.STRICT) {
750-
throw new UnknownSessionException("Unknown session " + sessionId);
751-
}
752711
sessions.putIfAbsent(sessionId, SessionState.PENDING_LOAD);
753712
sessionStream(sessionId);
754713
}
@@ -776,15 +735,11 @@ private void validateClientResponseScope(AcpSchema.JSONRPCResponse response, Str
776735
private OutboundStream openSessionStream(String sessionId) {
777736
SessionState current = sessions.get(sessionId);
778737
if (current == null) {
779-
if (routingMode == RoutingMode.STRICT) {
780-
throw new UnknownSessionException("Unknown session " + sessionId);
781-
}
782738
/*
783739
* RFD gap:
784740
* The current text says unknown session-scoped GET requests return 404,
785741
* but its resume flow also asks clients to open a session stream before
786-
* sending session/load. Compatible mode keeps a provisional stream so
787-
* practical resume can work while strict mode preserves the literal rule.
742+
* sending session/load. Keep a provisional stream so practical resume can work.
788743
*/
789744
sessions.putIfAbsent(sessionId, SessionState.PENDING_LOAD);
790745
}

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

Lines changed: 14 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ class StreamableHttpAcpAgentTransportIntegrationTest {
4949

5050
@Test
5151
void javaClientCanTalkToRunningJavaServer() throws Exception {
52-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
52+
try (FixtureServer server = FixtureServer.start()) {
5353
AcpAsyncClient client = AcpClient
5454
.async(new StreamableHttpAcpClientTransport(server.endpoint(), AcpJsonMapper.createDefault()))
5555
.requestTimeout(TIMEOUT)
@@ -73,7 +73,7 @@ void javaClientCanTalkToRunningJavaServer() throws Exception {
7373

7474
@Test
7575
void permissionRequestRoundTripsOverSessionStream() throws Exception {
76-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
76+
try (FixtureServer server = FixtureServer.start()) {
7777
AtomicInteger permissionRequests = new AtomicInteger();
7878
AcpAsyncClient client = AcpClient
7979
.async(new StreamableHttpAcpClientTransport(server.endpoint(), AcpJsonMapper.createDefault()))
@@ -103,7 +103,7 @@ void permissionRequestRoundTripsOverSessionStream() throws Exception {
103103

104104
@Test
105105
void compatibleModeAllowsSessionLoadPreopen() throws Exception {
106-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
106+
try (FixtureServer server = FixtureServer.start()) {
107107
AcpAsyncClient client = AcpClient
108108
.async(new StreamableHttpAcpClientTransport(server.endpoint(), AcpJsonMapper.createDefault()))
109109
.requestTimeout(TIMEOUT)
@@ -122,7 +122,7 @@ void compatibleModeAllowsSessionLoadPreopen() throws Exception {
122122

123123
@Test
124124
void supportsTwoLogicalSessions() throws Exception {
125-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
125+
try (FixtureServer server = FixtureServer.start()) {
126126
AcpAsyncClient client = AcpClient
127127
.async(new StreamableHttpAcpClientTransport(server.endpoint(), AcpJsonMapper.createDefault()))
128128
.requestTimeout(TIMEOUT)
@@ -153,7 +153,7 @@ void supportsTwoLogicalSessions() throws Exception {
153153

154154
@Test
155155
void wrongStreamClientResponseIsRejected() throws Exception {
156-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
156+
try (FixtureServer server = FixtureServer.start()) {
157157
HttpClient rawClient = HttpClient.newHttpClient();
158158
String connectionId = initializeRaw(rawClient, server.endpoint());
159159
try (SseReader connectionStream = SseReader.open(rawClient, server.endpoint(), connectionId, null)) {
@@ -187,7 +187,7 @@ void wrongStreamClientResponseIsRejected() throws Exception {
187187

188188
@Test
189189
void validationFailuresUseHttpStatusCodes() throws Exception {
190-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
190+
try (FixtureServer server = FixtureServer.start()) {
191191
HttpClient rawClient = HttpClient.newHttpClient();
192192
HttpResponse<String> jsonWithCharset = rawClient.send(HttpRequest.newBuilder(server.endpoint())
193193
.header("Content-Type", "application/json; charset=utf-8")
@@ -228,7 +228,7 @@ void validationFailuresUseHttpStatusCodes() throws Exception {
228228

229229
@Test
230230
void connectionReplayDeliversSessionNewWhenSseAttachesAfterPost() throws Exception {
231-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
231+
try (FixtureServer server = FixtureServer.start()) {
232232
HttpClient rawClient = HttpClient.newHttpClient();
233233
String connectionId = initializeRaw(rawClient, server.endpoint());
234234

@@ -251,7 +251,7 @@ void connectionReplayDeliversSessionNewWhenSseAttachesAfterPost() throws Excepti
251251

252252
@Test
253253
void sessionReplayDeliversPromptEventsWhenSseAttachesAfterPost() throws Exception {
254-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
254+
try (FixtureServer server = FixtureServer.start()) {
255255
HttpClient rawClient = HttpClient.newHttpClient();
256256
String connectionId = initializeRaw(rawClient, server.endpoint());
257257
String sessionId = createSession(rawClient, server.endpoint(), connectionId);
@@ -275,7 +275,7 @@ void sessionReplayDeliversPromptEventsWhenSseAttachesAfterPost() throws Exceptio
275275

276276
@Test
277277
void concurrentPostsToSameConnectionAreBothAcceptedAndRouted() throws Exception {
278-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
278+
try (FixtureServer server = FixtureServer.start()) {
279279
HttpClient rawClient = HttpClient.newHttpClient();
280280
String connectionId = initializeRaw(rawClient, server.endpoint());
281281
ExecutorService executor = Executors.newFixedThreadPool(2);
@@ -313,7 +313,7 @@ void concurrentPostsToSameConnectionAreBothAcceptedAndRouted() throws Exception
313313

314314
@Test
315315
void sessionScopedMessagesValidateSessionHeader() throws Exception {
316-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
316+
try (FixtureServer server = FixtureServer.start()) {
317317
HttpClient rawClient = HttpClient.newHttpClient();
318318
String connectionId = initializeRaw(rawClient, server.endpoint());
319319
String sessionId = createSession(rawClient, server.endpoint(), connectionId);
@@ -350,7 +350,7 @@ void sessionScopedMessagesValidateSessionHeader() throws Exception {
350350

351351
@Test
352352
void deleteClosesSseAndRemovesConnection() throws Exception {
353-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
353+
try (FixtureServer server = FixtureServer.start()) {
354354
HttpClient rawClient = HttpClient.newHttpClient();
355355
String connectionId = initializeRaw(rawClient, server.endpoint());
356356
try (SseReader connectionStream = SseReader.open(rawClient, server.endpoint(), connectionId, null)) {
@@ -385,7 +385,7 @@ void deleteClosesSseAndRemovesConnection() throws Exception {
385385

386386
@Test
387387
void replayOverflowClosesConnectionInsteadOfDroppingMessages() throws Exception {
388-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.COMPATIBLE)) {
388+
try (FixtureServer server = FixtureServer.start()) {
389389
HttpClient rawClient = HttpClient.newHttpClient();
390390
String connectionId = initializeRaw(rawClient, server.endpoint());
391391

@@ -404,31 +404,6 @@ void replayOverflowClosesConnectionInsteadOfDroppingMessages() throws Exception
404404
}
405405
}
406406

407-
@Test
408-
void strictModeRejectsUnknownSessionStream() throws Exception {
409-
try (FixtureServer server = FixtureServer.start(StreamableHttpAcpAgentTransport.RoutingMode.STRICT)) {
410-
HttpResponse<Void> response = HttpClient.newHttpClient()
411-
.send(HttpRequest.newBuilder(server.endpoint())
412-
.header("Content-Type", "application/json")
413-
.header("Accept", "application/json")
414-
.POST(HttpRequest.BodyPublishers.ofString("""
415-
{"jsonrpc":"2.0","id":"init-1","method":"initialize","params":{"protocolVersion":1,"clientCapabilities":{}}}
416-
"""))
417-
.build(), HttpResponse.BodyHandlers.discarding());
418-
String connectionId = response.headers().firstValue("Acp-Connection-Id").orElseThrow();
419-
HttpResponse<Void> unknownSession = HttpClient.newHttpClient()
420-
.send(HttpRequest.newBuilder(server.endpoint())
421-
.header("Accept", "text/event-stream")
422-
.header("Acp-Connection-Id", connectionId)
423-
.header("Acp-Session-Id", "unknown")
424-
.GET()
425-
.build(), HttpResponse.BodyHandlers.discarding());
426-
427-
assertThat(response.statusCode()).isEqualTo(200);
428-
assertThat(unknownSession.statusCode()).isEqualTo(404);
429-
}
430-
}
431-
432407
private static String initializeRaw(HttpClient client, URI endpoint) throws Exception {
433408
HttpResponse<String> initialize = client.send(HttpRequest.newBuilder(endpoint)
434409
.header("Content-Type", "application/json")
@@ -494,7 +469,7 @@ private FixtureServer(StreamableHttpAcpAgentTransport transport) {
494469
this.transport = transport;
495470
}
496471

497-
static FixtureServer start(StreamableHttpAcpAgentTransport.RoutingMode routingMode) throws Exception {
472+
static FixtureServer start() throws Exception {
498473
AtomicInteger sessionCounter = new AtomicInteger();
499474
AcpAgentFactory agentFactory = AcpAgentFactory.async(transport -> AcpAgent.async(transport)
500475
.initializeHandler(request -> Mono.just(new AcpSchema.InitializeResponse(
@@ -511,7 +486,7 @@ static FixtureServer start(StreamableHttpAcpAgentTransport.RoutingMode routingMo
511486
})
512487
.build());
513488
StreamableHttpAcpAgentTransport transport = new StreamableHttpAcpAgentTransport(
514-
freePort(), AcpJsonMapper.createDefault(), agentFactory).routingMode(routingMode);
489+
freePort(), AcpJsonMapper.createDefault(), agentFactory);
515490
transport.start().block(TIMEOUT);
516491
return new FixtureServer(transport);
517492
}

0 commit comments

Comments
 (0)