fix: call tasks/result to deliver side-channel messages

Replace polling loop in requestStream() with direct tasks/result call
that delivers queued messages (elicitation, sampling) via SSE.
Add integration test for callToolStream with elicitation.

Author: felixweinberger

src/integration-tests/taskLifecycle.test.ts

@@ -1604,4 +1604,77 @@ describe('Task Lifecycle Integration Tests', () => {
             await transport.close();
         });
     });
+
+    describe('callToolStream with elicitation', () => {
+        it('should deliver elicitation via callToolStream and complete task', async () => {
+            const client = new Client(
+                {
+                    name: 'test-client',
+                    version: '1.0.0'
+                },
+                {
+                    capabilities: {
+                        elicitation: {}
+                    }
+                }
+            );
+
+            // Track elicitation request receipt
+            let elicitationReceived = false;
+            let elicitationMessage = '';
+
+            // Set up elicitation handler on client
+            client.setRequestHandler(ElicitRequestSchema, async request => {
+                elicitationReceived = true;
+                elicitationMessage = request.params.message;
+
+                return {
+                    action: 'accept' as const,
+                    content: {
+                        userName: 'StreamUser'
+                    }
+                };
+            });
+
+            const transport = new StreamableHTTPClientTransport(baseUrl);
+            await client.connect(transport);
+
+            // Use callToolStream instead of raw request()
+            const stream = client.experimental.tasks.callToolStream({ name: 'input-task', arguments: {} }, CallToolResultSchema, {
+                task: { ttl: 60000 }
+            });
+
+            // Collect all stream messages
+            const messages: Array<{ type: string; task?: unknown; result?: unknown; error?: unknown }> = [];
+            for await (const message of stream) {
+                messages.push(message);
+            }
+
+            // Verify stream yielded expected message types
+            expect(messages.length).toBeGreaterThanOrEqual(2);
+
+            // First message should be taskCreated
+            expect(messages[0].type).toBe('taskCreated');
+            expect(messages[0].task).toBeDefined();
+
+            // Should have a taskStatus message
+            const statusMessages = messages.filter(m => m.type === 'taskStatus');
+            expect(statusMessages.length).toBeGreaterThanOrEqual(1);
+
+            // Last message should be result
+            const lastMessage = messages[messages.length - 1];
+            expect(lastMessage.type).toBe('result');
+            expect(lastMessage.result).toBeDefined();
+
+            // Verify elicitation was received and processed
+            expect(elicitationReceived).toBe(true);
+            expect(elicitationMessage).toContain('What is your name?');
+
+            // Verify result content
+            const result = lastMessage.result as { content: Array<{ type: string; text: string }> };
+            expect(result.content).toEqual([{ type: 'text', text: 'Hello, StreamUser!' }]);
+
+            await transport.close();
+        }, 15000);
+    });
 });

src/shared/protocol.ts

@@ -995,39 +995,35 @@ export abstract class Protocol<SendRequestT extends Request, SendNotificationT e
                 throw new McpError(ErrorCode.InternalError, 'Task creation did not return a task');
             }
 
-            // Poll for task completion
-            while (true) {
-                // Get current task status
-                const task = await this.getTask({ taskId }, options);
-                yield { type: 'taskStatus', task };
-
-                // Check if task is terminal
-                if (isTerminal(task.status)) {
-                    if (task.status === 'completed') {
-                        // Get the final result
-                        const result = await this.getTaskResult({ taskId }, resultSchema, options);
-                        yield { type: 'result', result };
-                    } else if (task.status === 'failed') {
-                        yield {
-                            type: 'error',
-                            error: new McpError(ErrorCode.InternalError, `Task ${taskId} failed`)
-                        };
-                    } else if (task.status === 'cancelled') {
-                        yield {
-                            type: 'error',
-                            error: new McpError(ErrorCode.InternalError, `Task ${taskId} was cancelled`)
-                        };
-                    }
-                    return;
+            // Get initial task status
+            const initialTask = await this.getTask({ taskId }, options);
+            yield { type: 'taskStatus', task: initialTask };
+
+            // If already terminal, handle immediately
+            if (isTerminal(initialTask.status)) {
+                if (initialTask.status === 'completed') {
+                    const result = await this.getTaskResult({ taskId }, resultSchema, options);
+                    yield { type: 'result', result };
+                } else if (initialTask.status === 'failed') {
+                    yield {
+                        type: 'error',
+                        error: new McpError(ErrorCode.InternalError, `Task ${taskId} failed`)
+                    };
+                } else if (initialTask.status === 'cancelled') {
+                    yield {
+                        type: 'error',
+                        error: new McpError(ErrorCode.InternalError, `Task ${taskId} was cancelled`)
+                    };
                 }
-
-                // Wait before polling again
-                const pollInterval = task.pollInterval ?? this._options?.defaultTaskPollInterval ?? 1000;
-                await new Promise(resolve => setTimeout(resolve, pollInterval));
-
-                // Check if cancelled
-                options?.signal?.throwIfAborted();
+                return;
             }
+
+            // Call tasks/result - this will:
+            // 1. Deliver any queued messages (elicitation, sampling) via SSE
+            // 2. Block until task becomes terminal
+            // 3. Return final result
+            const result = await this.getTaskResult({ taskId }, resultSchema, options);
+            yield { type: 'result', result };
         } catch (error) {
             yield {
                 type: 'error',