From 89d2a8236992842ba0f961a77c1a7e89e3875a68 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 00:41:31 +0000 Subject: [PATCH 1/2] fix(binding-mcp): defer mcp(client) lifecycle connect until guard decides McpLifecycleStream -- the one persistent south connection an mcp(client) binding multiplexes every subsequent method call over -- inherited the base McpStream#proceedWithRequest, which resolves the configured guard synchronously and connects regardless of the outcome. A guard whose authorization decision is asynchronous, or needs an interactive step (GuardHandler#NEEDS_PREAUTHORIZE), never actually got the chance to gate or challenge this connection: proceedWithRequest returned true before the guard had decided anything. McpLifecycleStream now overrides proceedWithRequest to defer the actual connect until guard.reauthorize's completion callback fires, mirroring the pattern already used for the per-request McpRequestStream. A NEEDS_PREAUTHORIZE decision is surfaced as an elicitCreate challenge -- the same challenge already sent for a backend-initiated elicitation -- so the existing north-facing challenge relay carries it through unchanged; the login callback answer resumes the deferred connect through the existing onReauthorized handling. Resuming also grants the app-level window, matching McpRequestStream#onAcquireCompleted, since omitting it left the connection timing out and reconnecting in a loop. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01W7N8Z9vABdU8pVPGthRodY --- .../mcp/internal/stream/McpClientFactory.java | 173 ++++++++++++++++++ 1 file changed, 173 insertions(+) diff --git a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java index 32227a340a..42d3ecee88 100644 --- a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java +++ b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java @@ -2343,6 +2343,17 @@ private final class McpLifecycleStream extends McpStream private HttpEventStream sse; boolean eventsUnsupported; + // guard-gated connect state: distinct from elicitCorrelationId/reauthTraceId above, + // which answer a backend-initiated elicitation on an already-open connection. These + // instead defer the very first doEncodeRequestBegin/onAppBeginImpl below until the + // guard actually decides -- see proceedWithRequest, which the base McpStream always + // resolves synchronously and unconditionally proceeds regardless of the outcome + private boolean pendingConnect; + private long acquireTraceId; + private long acquireAuthorization; + private String guardElicitElicitationId; + private String guardElicitCorrelationId; + private final LongCompletionCallback reauthorizeCompletion = new LongCompletionCallback() { @Override @@ -2361,12 +2372,43 @@ public void failed( } }; + private final LongCompletionCallback connectAcquireCompletion = new LongCompletionCallback() + { + @Override + public void completed( + long contextId, + long sessionId) + { + if (pendingConnect) + { + onConnectAcquireDecided(sessionId); + } + } + + @Override + public void failed( + long contextId, + Throwable ex) + { + if (pendingConnect) + { + onConnectAcquireDecided(GuardHandler.NOT_AUTHORIZED); + } + } + }; + @Override McpBindingConfig binding() { return binding; } + @Override + boolean awaitingAuth() + { + return pendingConnect; + } + @Override String transportSessionId() { @@ -2443,6 +2485,122 @@ boolean unregister( return requests.remove(id) != null; } + /** + * Unlike the base {@link McpStream#proceedWithRequest}, which always resolves the + * guard synchronously and proceeds regardless of the outcome, this defers the actual + * connect (below, via {@link #resumeDeferredConnect}) until the guard has genuinely + * decided -- including the async, potentially-interactive path a DCR + authorization- + * code guard needs. A {@code NEEDS_PREAUTHORIZE} decision is surfaced as the same + * {@code elicitCreate} challenge {@link #onDecodeElicitCreate} already sends for a + * backend-initiated elicitation, so the existing relay in McpProxyLifecycleFactory + * (settling this client's contribution to session establishment, then forwarding the + * challenge north) carries it through unchanged; the answer arrives the same way too, + * on {@link #onAppFlush}'s existing {@code KIND_ELICIT_CALLBACK} handling. + */ + @Override + boolean proceedWithRequest( + long traceId, + long authorization, + McpBeginExFW mcpBeginEx) + { + final GuardHandler guard = binding.guard; + if (guard == null) + { + return true; + } + + if ((authorization & GuardHandler.MASK_AUTHORIZED) != 0L) + { + final String resolved = guard.credentials(authorization); + if (resolved != null) + { + credentials = resolved; + return true; + } + } + + acquireTraceId = traceId; + acquireAuthorization = authorization; + pendingConnect = true; + guard.reauthorize(traceId, binding.id, authorization, null, connectAcquireCompletion); + + return false; + } + + private void onConnectAcquireDecided( + long sessionId) + { + final GuardHandler guard = binding.guard; + + if ((sessionId & GuardHandler.MASK_AUTHORIZED) != 0L) + { + pendingConnect = false; + credentials = guard.credentials(sessionId); + guardSessionId = sessionId; + resumeDeferredConnect(acquireTraceId, acquireAuthorization); + return; + } + + if (sessionId == GuardHandler.NEEDS_PREAUTHORIZE && binding.needsCredentials) + { + final String preauthorizeUrl = + guard.preauthorize(acquireTraceId, binding.id, initialId, acquireAuthorization, authCallback); + if (preauthorizeUrl == null) + { + pendingConnect = false; + doAppReset(acquireTraceId, acquireAuthorization); + doAppAbort(acquireTraceId, acquireAuthorization); + return; + } + + guardElicitElicitationId = supplyElicitationId.get(); + guardElicitCorrelationId = supplyElicitCorrelationId.get(); + + final McpChallengeExFW challengeEx = mcpChallengeExRW + .wrap(extBuffer, 0, extBuffer.capacity()) + .typeId(mcpTypeId) + .elicitCreate(b -> b + .id(guardElicitElicitationId) + .url(preauthorizeUrl) + .message(ELICIT_MESSAGE_PREAUTHORIZE) + .correlationId(guardElicitCorrelationId)) + .build(); + doAppChallenge(acquireTraceId, acquireAuthorization, challengeEx); + + // pendingConnect stays true: the decision -- and the deferred connect it + // gates -- resumes from onReauthorized once the login callback answers it + return; + } + + pendingConnect = false; + if (!binding.needsCredentials) + { + resumeDeferredConnect(acquireTraceId, acquireAuthorization); + } + else + { + doAppReset(acquireTraceId, acquireAuthorization); + doAppAbort(acquireTraceId, acquireAuthorization); + } + } + + /** + * Mirrors {@link McpRequestStream#onAcquireCompleted}: {@link #onAppBegin} withheld + * the app-level window while {@code awaitingAuth()} (via {@link #pendingConnect}), so + * granting it here is what lets whoever is waiting on this lifecycle connection's own + * accept stop waiting -- omitting it left the caller timing out and tearing the + * connection down long after the guard had actually decided, in an endless reconnect + * loop, since nothing else ever re-arms the window this class deferred. + */ + private void resumeDeferredConnect( + long traceId, + long authorization) + { + http.doEncodeRequestBegin(traceId, authorization); + onAppBeginImpl(traceId, authorization, null); + doAppWindow(traceId, authorization, 0L, 0); + } + @Override void onAppBeginImpl( long traceId, @@ -2701,6 +2859,13 @@ private void onReauthorized( deauthorizeGuardSession(); guardSessionId = sessionId; + if (pendingConnect) + { + pendingConnect = false; + resumeDeferredConnect(acquireTraceId, acquireAuthorization); + return; + } + final McpFlushExFW flushEx = mcpFlushExRW .wrap(extBuffer, 0, extBuffer.capacity()) .typeId(mcpTypeId) @@ -2710,6 +2875,14 @@ private void onReauthorized( .build(); doAppFlush(reauthTraceId, reauthAuthorization, flushEx); } + else if (pendingConnect) + { + // the login callback failed to resolve to a usable session -- the connect + // this client deferred in proceedWithRequest has nothing left to wait for + pendingConnect = false; + doAppReset(reauthTraceId, reauthAuthorization); + doAppAbort(reauthTraceId, reauthAuthorization); + } } @Override From 92cb8e8eaf7fbc484b1bd79eb0d4e192d9162b08 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 00:48:43 +0000 Subject: [PATCH 2/2] style(binding-mcp): trim explanatory comments from the lifecycle connect fix Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01W7N8Z9vABdU8pVPGthRodY --- .../mcp/internal/stream/McpClientFactory.java | 29 ------------------- 1 file changed, 29 deletions(-) diff --git a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java index 42d3ecee88..5565044089 100644 --- a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java +++ b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/stream/McpClientFactory.java @@ -2343,11 +2343,6 @@ private final class McpLifecycleStream extends McpStream private HttpEventStream sse; boolean eventsUnsupported; - // guard-gated connect state: distinct from elicitCorrelationId/reauthTraceId above, - // which answer a backend-initiated elicitation on an already-open connection. These - // instead defer the very first doEncodeRequestBegin/onAppBeginImpl below until the - // guard actually decides -- see proceedWithRequest, which the base McpStream always - // resolves synchronously and unconditionally proceeds regardless of the outcome private boolean pendingConnect; private long acquireTraceId; private long acquireAuthorization; @@ -2485,18 +2480,6 @@ boolean unregister( return requests.remove(id) != null; } - /** - * Unlike the base {@link McpStream#proceedWithRequest}, which always resolves the - * guard synchronously and proceeds regardless of the outcome, this defers the actual - * connect (below, via {@link #resumeDeferredConnect}) until the guard has genuinely - * decided -- including the async, potentially-interactive path a DCR + authorization- - * code guard needs. A {@code NEEDS_PREAUTHORIZE} decision is surfaced as the same - * {@code elicitCreate} challenge {@link #onDecodeElicitCreate} already sends for a - * backend-initiated elicitation, so the existing relay in McpProxyLifecycleFactory - * (settling this client's contribution to session establishment, then forwarding the - * challenge north) carries it through unchanged; the answer arrives the same way too, - * on {@link #onAppFlush}'s existing {@code KIND_ELICIT_CALLBACK} handling. - */ @Override boolean proceedWithRequest( long traceId, @@ -2567,8 +2550,6 @@ private void onConnectAcquireDecided( .build(); doAppChallenge(acquireTraceId, acquireAuthorization, challengeEx); - // pendingConnect stays true: the decision -- and the deferred connect it - // gates -- resumes from onReauthorized once the login callback answers it return; } @@ -2584,14 +2565,6 @@ private void onConnectAcquireDecided( } } - /** - * Mirrors {@link McpRequestStream#onAcquireCompleted}: {@link #onAppBegin} withheld - * the app-level window while {@code awaitingAuth()} (via {@link #pendingConnect}), so - * granting it here is what lets whoever is waiting on this lifecycle connection's own - * accept stop waiting -- omitting it left the caller timing out and tearing the - * connection down long after the guard had actually decided, in an endless reconnect - * loop, since nothing else ever re-arms the window this class deferred. - */ private void resumeDeferredConnect( long traceId, long authorization) @@ -2877,8 +2850,6 @@ private void onReauthorized( } else if (pendingConnect) { - // the login callback failed to resolve to a usable session -- the connect - // this client deferred in proceedWithRequest has nothing left to wait for pendingConnect = false; doAppReset(reauthTraceId, reauthAuthorization); doAppAbort(reauthTraceId, reauthAuthorization);