diff --git a/.changeset/progress-notification-same-read.md b/.changeset/progress-notification-same-read.md new file mode 100644 index 0000000000..07190da71b --- /dev/null +++ b/.changeset/progress-notification-same-read.md @@ -0,0 +1,6 @@ +--- +'@modelcontextprotocol/client': patch +'@modelcontextprotocol/server': patch +--- + +Progress notifications that arrive in the same transport read as their request's response are now delivered to the `onprogress` callback before the response settles the request. Previously the notification handler ran in a microtask after the response had already removed the progress handler, so those updates were silently dropped. Only `notifications/progress` is dispatched synchronously; other notifications keep microtask dispatch. diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index ea86863a4f..c9074ea819 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -959,10 +959,21 @@ export abstract class Protocol { extra ); - // Starting with Promise.resolve() puts any synchronous errors into the monad as well. - runInScope(scope, () => - Promise.resolve().then(() => (handler === undefined ? fallback!(notification) : handler(notification, codec))) - ).then( + const dispatch = () => (handler === undefined ? fallback!(notification) : handler(notification, codec)); + + // Progress is dispatched synchronously so a progress notification that + // shares a read with its request's response is not deferred past + // _onresponse, which deletes the matching progress handler (see #3018). + // Other notifications keep microtask dispatch so their handlers do not + // run inside the transport's onmessage stack. Either way, synchronous + // throws and async rejections both settle the promise below and route to _onerror. + const settled = runInScope(scope, () => + notification.method === 'notifications/progress' + ? new Promise(resolve => resolve(dispatch())) + : Promise.resolve().then(dispatch) + ); + + settled.then( () => scope?.end({ type: 'delivered' }), error => { scope?.end({ type: 'failed', error }); diff --git a/packages/core-internal/test/shared/protocol.test.ts b/packages/core-internal/test/shared/protocol.test.ts index 2fb0f64813..47649a4bff 100644 --- a/packages/core-internal/test/shared/protocol.test.ts +++ b/packages/core-internal/test/shared/protocol.test.ts @@ -617,6 +617,85 @@ describe('protocol tests', () => { await Promise.resolve(); await expect(requestPromise).resolves.toEqual({ result: 'success' }); }); + + test('should deliver progress when notification and response arrive in the same turn', async () => { + // Repro for #3018: progress was deferred via Promise.resolve(), so + // a same-turn response deleted the progress handler first. + await protocol.connect(transport); + const request = { method: 'example', params: {} }; + const mockSchema: ZodType<{ result: string }> = z.object({ + result: z.string() + }); + const onProgressMock = vi.fn(); + const onErrorMock = vi.fn(); + protocol.onerror = onErrorMock; + + const requestPromise = testRequest(protocol, request, mockSchema, { + timeout: 1000, + onprogress: onProgressMock + }); + + if (transport.onmessage) { + transport.onmessage({ + jsonrpc: '2.0', + method: 'notifications/progress', + params: { + progressToken: 0, + progress: 50, + total: 100 + } + }); + transport.onmessage({ + jsonrpc: '2.0', + id: 0, + result: { result: 'success' } + }); + } + + await expect(requestPromise).resolves.toEqual({ result: 'success' }); + expect(onProgressMock).toHaveBeenCalledWith({ + progress: 50, + total: 100 + }); + expect(onErrorMock).not.toHaveBeenCalled(); + }); + }); + + describe('Notification dispatch ordering', () => { + test('keeps non-progress notification handlers on microtask dispatch', async () => { + await protocol.connect(transport); + const order: string[] = []; + protocol.fallbackNotificationHandler = async () => { + order.push('handler'); + }; + + transport.onmessage?.({ jsonrpc: '2.0', method: 'notifications/custom-extension', params: {} }); + order.push('after-onmessage'); + + await Promise.resolve(); + await Promise.resolve(); + expect(order).toEqual(['after-onmessage', 'handler']); + }); + + test('a non-progress handler that closes the connection does not run inside onmessage', async () => { + await protocol.connect(transport); + const onErrorMock = vi.fn(); + protocol.onerror = onErrorMock; + let closedDuringOnmessage = false; + let inOnmessage = false; + protocol.fallbackNotificationHandler = async () => { + closedDuringOnmessage = inOnmessage; + await protocol.close(); + }; + + inOnmessage = true; + transport.onmessage?.({ jsonrpc: '2.0', method: 'notifications/custom-extension', params: {} }); + inOnmessage = false; + + await vi.waitFor(() => expect(protocol.transport).toBeUndefined()); + expect(closedDuringOnmessage).toBe(false); + expect(onErrorMock).not.toHaveBeenCalled(); + }); }); describe('Debounced Notifications', () => { diff --git a/test/e2e/requirements.ts b/test/e2e/requirements.ts index d26f6524d2..d84fbc7b6d 100644 --- a/test/e2e/requirements.ts +++ b/test/e2e/requirements.ts @@ -218,23 +218,11 @@ export const REQUIREMENTS: Record = { 'protocol:progress:callback': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/progress#progress-flow', behavior: - "Progress notifications emitted by a handler during a request are delivered to the caller's progress callback, in order, with their progress, total, and message.", - knownFailures: [ - { - transport: 'sse', - note: "Real-socket SSE delivers a handler's progress notifications and its response in one batch; the response is processed first, so the progress notifications never reach the caller's progress callback." - } - ] + "Progress notifications emitted by a handler during a request are delivered to the caller's progress callback, in order, with their progress, total, and message." }, 'typescript:protocol:progress:token-injected': { source: 'sdk', - behavior: 'Passing onprogress causes a progressToken to be injected into request _meta, preserving existing _meta fields.', - knownFailures: [ - { - transport: 'sse', - note: "Real-socket SSE delivers a handler's progress notifications and its response in one batch; the response is processed first, so the progress notifications never reach the caller's progress callback." - } - ] + behavior: 'Passing onprogress causes a progressToken to be injected into request _meta, preserving existing _meta fields.' }, 'protocol:progress:token-unique': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/progress#progress-flow', @@ -250,13 +238,7 @@ export const REQUIREMENTS: Record = { }, 'protocol:timeout:reset-on-progress': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#timeouts', - behavior: "When configured to do so, each progress notification resets the request's read timeout.", - knownFailures: [ - { - transport: 'sse', - note: 'Same real-socket SSE batching race as protocol:progress:callback: the progress notifications are dropped before they can reset the timeout, so the request times out.' - } - ] + behavior: "When configured to do so, each progress notification resets the request's read timeout." }, 'protocol:timeout:sends-cancellation': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#timeouts', @@ -395,13 +377,7 @@ export const REQUIREMENTS: Record = { }, 'tools:call:progress': { source: 'https://modelcontextprotocol.io/specification/2025-11-25/basic/utilities/progress#progress-flow', - behavior: "Progress notifications emitted by a tool handler reach the caller's progress callback before the tool result returns.", - knownFailures: [ - { - transport: 'sse', - note: "Real-socket SSE delivers a handler's progress notifications and its response in one batch; the response is processed first, so the progress notifications never reach the caller's progress callback." - } - ] + behavior: "Progress notifications emitted by a tool handler reach the caller's progress callback before the tool result returns." }, 'tools:call:sampling-roundtrip': { transports: STATEFUL_TRANSPORTS,