Expose actual inbound-processing completion for reusable transport admission

.changeset/inbound-processing-completion.md

@@ -0,0 +1,8 @@
+---
+'@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. Completion remains independent of OpenTelemetry span outcomes and permits request-ID reuse from the callback after controller cleanup.
+
+Queue subscription acknowledgements before routed changes and retain pending writes without blocking unrelated cancellation. Client completion also waits for asynchronous cache invalidation while notification handlers dispatch immediately.

docs/inbound-processing-completion.md

@@ -0,0 +1,11 @@
+# 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 queued before routed change notifications and is not completion. The stdio entry waits for cancellation or teardown, including a pending acknowledgement and any graceful terminal write it owns. Pending subscription writes retain admission without blocking unrelated inbound cancellation. `supportsInboundMessageCompletion` is exported from `@modelcontextprotocol/server/stdio` for adapters that require this contract before accepting input.
+
+Client change notifications include asynchronous cache eviction in completion. Notification handlers start immediately while the cache store is pending; completion waits for both the handler and eviction from all affected cache partitions.
+
+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.
+
+Completion is independent of OpenTelemetry span outcomes: a span may end before a response write settles or when cancellation starts, while the message remains admitted until its owned work finishes. Completion is not FIFO across messages. An adapter may synchronously reuse a completed request ID from its callback; the previous request controller has already been removed.

packages/client/src/client/client.ts

@@ -2206,11 +2206,12 @@ 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> {
+        const evictions: 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
-        // entry. Runs regardless of whether the user
+        // matching cached list result is stale. Start eviction (do NOT refetch)
+        // before dispatch so synchronous stores are cleared when a handler runs.
+        // Runs regardless of whether the user
         // configured `listChanged` — derived views (the tool index, output
         // validators) must drop the stale entry either way. `raw.method` is
         // server-controlled; guard with `Object.hasOwn` so an inherited
@@ -2222,10 +2223,9 @@ export class Client extends Protocol<ClientContext> {
             // `resources/read` body for this URI is stale on receipt; drop it
             // from BOTH partitions so the next `readResource` for the same URI
             // refetches even within TTL (the documented subscribe → updated →
-            // re-read flow). Fire-and-forget like the `list_changed` path —
-            // dispatch must not block on an async store.
+            // re-read flow). Start eviction before dispatch and join both below.
             const uri = (raw.params as { uri?: unknown } | undefined)?.uri;
-            if (typeof uri === 'string') void this._cache.evictKey('resources/read', uri);
+            if (typeof uri === 'string') evictions.push(this._cache.evictKey('resources/read', uri));
         } else if (evicted !== undefined) {
             for (const method of evicted) {
                 // `evict()` bumps the generation FIRST and unconditionally
@@ -2235,9 +2235,9 @@ export class Client extends Protocol<ClientContext> {
                 // store keep their entries. Store failures are reported via
                 // `onerror` inside `evict()` and the call resolves, so
                 // dispatch (and the user's `listChanged` handler) runs
-                // regardless. Fire-and-forget — dispatch must not block on
-                // an async store.
-                void this._cache.evict(method);
+                // regardless. Dispatch starts immediately; completion joins
+                // the asynchronous store work below.
+                evictions.push(this._cache.evict(method));
             }
         }
         if (raw.method === 'notifications/subscriptions/acknowledged') {
@@ -2266,7 +2266,8 @@ export class Client extends Protocol<ClientContext> {
                 return;
             }
         }
-        super._onnotification(raw, extra);
+        const handled = super._onnotification(raw, extra);
+        return evictions.length === 0 ? handled : Promise.all([...evictions, handled]).then(() => {});
     }
 
     /**
@@ -2279,7 +2280,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) {
@@ -2308,7 +2309,7 @@ export class Client extends Protocol<ClientContext> {
             }
             return;
         }
-        super._onresponse(response);
+        return super._onresponse(response);
     }
 
     /**

packages/client/test/client/responseCache.test.ts

@@ -7,9 +7,11 @@
  * `toolDefinition` hit/miss and re-derivation only on a stamp change; the
  * generation guard skipping a stale write.
  */
-import type { JSONRPCMessage, JSONRPCRequest, Tool } from '@modelcontextprotocol/core-internal';
+import { setImmediate as nextTurn } from 'node:timers/promises';
+
+import type { JSONRPCMessage, JSONRPCRequest, Tool, Transport } from '@modelcontextprotocol/core-internal';
 import { InMemoryTransport, SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal';
-import { describe, expect, it } from 'vitest';
+import { describe, expect, it, vi } from 'vitest';
 
 import { Client } from '../../src/client/client';
 import type { CacheEntry, ResponseCacheStore } from '../../src/client/responseCache';
@@ -422,6 +424,50 @@ const cacheOf = (client: Client): ClientResponseCache => (client as unknown as {
 const toolDef = (client: Client, name: string): Promise<Tool | undefined> => cacheOf(client).toolDefinition(name);
 
 describe('Client response-cache substrate', () => {
+    it.each([
+        { method: 'notifications/tools/list_changed', params: {}, deletesPerPartition: 1 },
+        { method: 'notifications/resources/list_changed', params: {}, deletesPerPartition: 2 },
+        { method: 'notifications/resources/updated', params: { uri: 'file:///a' }, deletesPerPartition: 1 }
+    ] as const)(
+        'completion joins async cache eviction for $method without delaying dispatch',
+        async ({ method, params, deletesPerPartition }) => {
+            const store = new InMemoryResponseCacheStore();
+            const pending: (() => void)[] = [];
+            const deleted: string[] = [];
+            (store as ResponseCacheStore).delete = async key => {
+                await new Promise<void>(resolve => pending.push(resolve));
+                deleted.push(key.partition!);
+            };
+            const transport: Transport = (await scriptedModernServer([])).clientTx;
+            const client = modernClient(store, { cachePartition: 'alice' });
+            const handled = vi.fn();
+            const complete = vi.fn();
+            client.setNotificationHandler(method, handled);
+            await client.connect(transport);
+            try {
+                transport.onmessage!({ jsonrpc: '2.0', method, params }, { onMessageProcessed: complete });
+                await nextTurn();
+                expect(handled).toHaveBeenCalledOnce();
+                expect(pending).toHaveLength(deletesPerPartition);
+                expect(complete).not.toHaveBeenCalled();
+                pending.splice(0).forEach(resolve => resolve());
+                await nextTurn();
+                expect(deleted).toEqual(Array(deletesPerPartition).fill(part('alice')));
+                expect(pending).toHaveLength(deletesPerPartition);
+                expect(complete).not.toHaveBeenCalled();
+                pending.splice(0).forEach(resolve => resolve());
+                await nextTurn();
+                expect(deleted).toEqual([...Array(deletesPerPartition).fill(part('alice')), ...Array(deletesPerPartition).fill(part())]);
+                expect(complete).toHaveBeenCalledOnce();
+            } finally {
+                pending.splice(0).forEach(resolve => resolve());
+                await nextTurn();
+                pending.splice(0).forEach(resolve => resolve());
+                await client.close();
+            }
+        }
+    );
+
     it('listTools() with no cursor reads every page, writes one cache entry; listTools({cursor}) stays per-page and does not write', async () => {
         const store = new InMemoryResponseCacheStore();
         const { clientTx, listCount } = await scriptedModernServer([[TOOL_A], [TOOL_B]]);