Expose actual inbound-processing completion for reusable transport admission

.changeset/inbound-processing-completion.md

@@ -0,0 +1,6 @@
+---
+'@modelcontextprotocol/server': minor
+'@modelcontextprotocol/client': minor
+---
+
+Add local inbound-processing completion metadata for transport adapters. Completion follows request handlers, response transmission and internal request cleanup; cancellation alone does not complete a request. The stdio entry preserves this metadata through opening and error paths and holds listen requests until subscription teardown. Export a stdio capability marker so reusable bounded admission can reject SDK versions without this contract.

docs/inbound-processing-completion.md

@@ -0,0 +1,7 @@
+# Local inbound-processing completion
+
+Transport adapters may attach `MessageExtraInfo.onMessageProcessed` when forwarding an inbound message to a connected protocol instance or the `serveStdio` entry. The callback runs once after SDK-owned processing, including asynchronous handlers, response `send()` settlement and request-controller cleanup. Failed handlers, failed writes and entry-level rejection paths also complete after their owned work. The official stdio transport joins the callback for each Node Writable chunk, including failed callbacks. Backpressure or stream error events alone do not complete that write. A cancellation notification or transport close does not settle an unfinished handler or write.
+
+For `subscriptions/listen`, the opening acknowledgement is not completion. The stdio entry waits for cancellation or teardown, including a pending acknowledgement and any graceful terminal write it owns. `supportsInboundMessageCompletion` is exported from `@modelcontextprotocol/server/stdio` for adapters that require this contract before accepting input.
+
+The metadata is local and is never serialized. It is suitable for returning a local admission credit or reusing an inbound request identifier. It does not prove delivery to the remote peer, remote effects, resource cleanup started outside the handler's returned promise, or application-owned physical settlement. Applications must join their own work before their handlers finish. Observers should not throw; observer errors are reported through the existing error callback.

packages/client/src/client/client.ts

@@ -2158,7 +2158,7 @@ export class Client extends Protocol<ClientContext> {
      * `setNotificationHandler` / `fallbackNotificationHandler` instead of
      * being silently swallowed.
      */
-    protected override _onnotification(raw: JSONRPCNotification, extra?: MessageExtraInfo): void {
+    protected override _onnotification(raw: JSONRPCNotification, extra?: MessageExtraInfo): void | Promise<void> {
         // Response-cache invalidation: a `list_changed` notification means the
         // matching cached list result is stale. Evict (do NOT refetch) before
         // dispatch so a handler that reaches the cache observes the cleared
@@ -2218,7 +2218,7 @@ export class Client extends Protocol<ClientContext> {
                 return;
             }
         }
-        super._onnotification(raw, extra);
+        return super._onnotification(raw, extra);
     }
 
     /**
@@ -2231,7 +2231,7 @@ export class Client extends Protocol<ClientContext> {
      * NUMBER and never holds a listen id, so passing a string-id response
      * through would surface as "unknown message ID" via `onerror`).
      */
-    protected override _onresponse(response: JSONRPCResponse): void {
+    protected override _onresponse(response: JSONRPCResponse): void | Promise<void> {
         const id = response.id;
         const entry = typeof id === 'string' ? this._listenState.get(id) : undefined;
         if (entry !== undefined) {
@@ -2260,7 +2260,7 @@ export class Client extends Protocol<ClientContext> {
             }
             return;
         }
-        super._onresponse(response);
+        return super._onresponse(response);
     }
 
     /**

packages/core-internal/src/shared/protocol.ts

@@ -570,7 +570,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
     private _customSchemaRequestMethods = new Set<string>();
     private _requestHandlerAbortControllers: Map<RequestId, AbortController> = new Map();
     private _notificationHandlers: Map<string, (notification: JSONRPCNotification, codec: WireCodec) => Promise<void>> = new Map();
-    private _responseHandlers: Map<number, (response: JSONRPCResultResponse | Error) => void> = new Map();
+    private _responseHandlers: Map<number, (response: JSONRPCResultResponse | Error) => void | Promise<void>> = new Map();
     private _progressHandlers: Map<number, ProgressCallback> = new Map();
     private _timeoutInfo: Map<number, TimeoutInfo> = new Map();
     private _pendingDebouncedNotifications = new Set<string>();
@@ -810,15 +810,48 @@ export abstract class Protocol<ContextT extends BaseContext> {
 
         const _onmessage = this._transport?.onmessage;
         this._transport.onmessage = (message, extra) => {
-            _onmessage?.(message, extra);
-            if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
-                this._onresponse(message);
-            } else if (isJSONRPCRequest(message)) {
-                this._onrequest(message, extra);
-            } else if (isJSONRPCNotification(message)) {
-                this._onnotification(message, extra);
-            } else {
-                this._onerror(new Error(`Unknown message type: ${JSON.stringify(message)}`));
+            let processed = false;
+            const complete = () => {
+                if (processed) return;
+                processed = true;
+                try {
+                    extra?.onMessageProcessed?.();
+                } catch (error) {
+                    this._onerror(error instanceof Error ? error : new Error('Inbound completion callback failed'));
+                }
+            };
+            let work: void | Promise<void> | undefined;
+            try {
+                _onmessage?.(message, extra);
+                if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
+                    work = this._onresponse(message);
+                } else if (isJSONRPCRequest(message)) {
+                    work = this._onrequest(message, extra);
+                } else if (isJSONRPCNotification(message)) {
+                    work = this._onnotification(message, extra);
+                } else {
+                    this._onerror(new Error(`Unknown message type: ${JSON.stringify(message)}`));
+                }
+            } catch (error) {
+                // Without a completion observer retain the historical dispatch behavior.
+                if (!extra?.onMessageProcessed) throw error;
+                this._onerror(error instanceof Error ? error : new Error('Inbound dispatch failed'));
+                work = isJSONRPCRequest(message)
+                    ? this._transport?.send({
+                          jsonrpc: '2.0',
+                          id: message.id,
+                          error: { code: ProtocolErrorCode.InternalError, message: 'Internal server error' }
+                      })
+                    : undefined;
+            }
+            if (extra?.onMessageProcessed) {
+                void Promise.resolve(work).then(complete, error => {
+                    try {
+                        this._onerror(error instanceof Error ? error : new Error('Inbound processing failed'));
+                    } finally {
+                        complete();
+                    }
+                });
             }
         };
 
@@ -874,7 +907,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
      * an override that consumes only what it owns and falls through to base
      * dispatch for everything else.
      */
-    protected _onnotification(rawNotification: JSONRPCNotification, extra?: MessageExtraInfo): void {
+    protected _onnotification(rawNotification: JSONRPCNotification, extra?: MessageExtraInfo): void | Promise<void> {
         // Hide wire-only material from notification handlers too — but ONLY
         // the reserved envelope `_meta` keys (the retry params names are
         // reserved on requests, not notifications). There is no
@@ -930,12 +963,12 @@ export abstract class Protocol<ContextT extends BaseContext> {
         }
 
         // Starting with Promise.resolve() puts any synchronous errors into the monad as well.
-        Promise.resolve()
+        return Promise.resolve()
             .then(() => (handler === undefined ? fallback!(notification) : handler(notification, codec)))
             .catch(error => this._onerror(new Error(`Uncaught error in notification handler: ${error}`)));
     }
 
-    private _onrequest(rawRequest: JSONRPCRequest, extra?: MessageExtraInfo): void {
+    private _onrequest(rawRequest: JSONRPCRequest, extra?: MessageExtraInfo): void | Promise<void> {
         // Lift wire-only material before dispatch: handlers (including the
         // fallback handler and the per-method schema parse) see exactly the
         // 2025-era shape; the envelope and retry fields surface via ctx.
@@ -967,7 +1000,9 @@ export abstract class Protocol<ContextT extends BaseContext> {
                 id: request.id,
                 error: { code, message, ...(data !== undefined && { data }) }
             };
-            capturedTransport?.send(errorResponse).catch(error => this._onerror(new Error(`Failed to send an error response: ${error}`)));
+            return capturedTransport
+                ?.send(errorResponse)
+                .catch(error => this._onerror(new Error(`Failed to send an error response: ${error}`)));
         };
 
         // Edge→instance handoff check: a classification that disagrees with
@@ -988,7 +1023,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
                 // only the fallback for classifications without an exact
                 // revision.
                 const requested = extra.classification.revision ?? classified;
-                sendErrorResponse(ProtocolErrorCode.UnsupportedProtocolVersion, `Unsupported protocol version: ${requested}`, {
+                return sendErrorResponse(ProtocolErrorCode.UnsupportedProtocolVersion, `Unsupported protocol version: ${requested}`, {
                     // Per spec, `supported` is the full list of protocol
                     // versions the receiver supports — not just the version
                     // this connection is on — so the peer can pick a mutually
@@ -999,7 +1034,6 @@ export abstract class Protocol<ContextT extends BaseContext> {
                     supported: this._supportedProtocolVersions,
                     requested
                 });
-                return;
             }
         }
 
@@ -1015,15 +1049,13 @@ export abstract class Protocol<ContextT extends BaseContext> {
             !codec.hasRequestMethod(request.method) &&
             !(isExtensionReusedRequestMethod(request.method) && this._customSchemaRequestMethods.has(request.method))
         ) {
-            sendErrorResponse(ProtocolErrorCode.MethodNotFound, 'Method not found');
-            return;
+            return sendErrorResponse(ProtocolErrorCode.MethodNotFound, 'Method not found');
         }
 
         const handler = this._requestHandlers.get(request.method) ?? this.fallbackRequestHandler;
 
         if (handler === undefined) {
-            sendErrorResponse(ProtocolErrorCode.MethodNotFound, 'Method not found');
-            return;
+            return sendErrorResponse(ProtocolErrorCode.MethodNotFound, 'Method not found');
         }
 
         // Envelope enforcement: the 2026 era requires the per-request `_meta`
@@ -1037,8 +1069,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
         // only the −32601-over-−32602 rule.)
         const envelopeError = codec.checkInboundEnvelope(lifted);
         if (envelopeError !== undefined) {
-            sendErrorResponse(ProtocolErrorCode.InvalidParams, envelopeError);
-            return;
+            return sendErrorResponse(ProtocolErrorCode.InvalidParams, envelopeError);
         }
 
         // Related sends resolve through the SAME instance era as every other
@@ -1108,11 +1139,10 @@ export abstract class Protocol<ContextT extends BaseContext> {
             },
             http: extra?.authInfo ? { authInfo: extra.authInfo } : undefined
         };
-        const ctx = this.buildContext(baseCtx, extra);
 
         // Starting with Promise.resolve() puts any synchronous errors into the monad as well.
-        Promise.resolve()
-            .then(() => handler(request, ctx))
+        return Promise.resolve()
+            .then(() => handler(request, this.buildContext(baseCtx, extra)))
             .then(
                 async result => {
                     if (abortController.signal.aborted) {
@@ -1133,8 +1163,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
                         encoded = codec.encodeResult(request.method, result, this._outboundServerInfo());
                     } catch (error) {
                         this._onerror(new Error(`Failed to encode result for ${request.method}: ${error}`));
-                        sendErrorResponse(ProtocolErrorCode.InternalError, 'Internal error');
-                        return;
+                        return sendErrorResponse(ProtocolErrorCode.InternalError, 'Internal error');
                     }
 
                     const response: JSONRPCResponse = {
@@ -1210,7 +1239,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
      * only what it owns and falls through to base dispatch for everything
      * else.
      */
-    protected _onresponse(response: JSONRPCResponse | JSONRPCErrorResponse): void {
+    protected _onresponse(response: JSONRPCResponse | JSONRPCErrorResponse): void | Promise<void> {
         const messageId = Number(response.id);
 
         const handler = this._responseHandlers.get(messageId);
@@ -1224,10 +1253,10 @@ export abstract class Protocol<ContextT extends BaseContext> {
         this._progressHandlers.delete(messageId);
 
         if (isJSONRPCResultResponse(response)) {
-            handler(response);
+            return handler(response);
         } else {
             const error = ProtocolError.fromError(response.error.code, response.error.message, response.error.data);
-            handler(error);
+            return handler(error);
         }
     }
 
@@ -1563,11 +1592,14 @@ export abstract class Protocol<ContextT extends BaseContext> {
                                 legOptions
                             )
                     };
-                    return resolve(this._resolveNonCompleteResult(decoded, flow) as Promise<StandardSchemaV1.InferOutput<T>>);
+                    return (this._resolveNonCompleteResult(decoded, flow) as Promise<StandardSchemaV1.InferOutput<T>>).then(
+                        resolve,
+                        reject
+                    );
                 }
                 const result = decoded.result;
 
-                validateStandardSchema(resultSchema, result).then(parseResult => {
+                return validateStandardSchema(resultSchema, result).then(parseResult => {
                     if (parseResult.success) {
                         resolve(parseResult.data);
                     } else {

packages/core-internal/src/types/types.ts

@@ -901,6 +901,17 @@ export interface MessageClassification {
  * Extra information about a message.
  */
 export interface MessageExtraInfo {
+    /**
+     * Local transport-side completion of this inbound message's SDK processing.
+     * Called once after dispatch, awaited response sends and request-controller
+     * cleanup, including cancellation with no response. For a stdio listen,
+     * completion waits for subscription teardown rather than its acknowledgement.
+     * This does not prove downstream remote effects or consumer-detached work
+     * have finished. Never serialized onto the wire. Callback errors are reported
+     * through the receiving protocol/entry's error handler.
+     */
+    onMessageProcessed?: () => void;
+
     /**
      * The original HTTP request.
      */