[Server] Shield a request handler from foreign fiber suspends

src/Server/Protocol.php

@@ -288,12 +288,14 @@ private function handleRequest(TransportInterface $transport, Request $request,
 
                 // One fiber for the whole exchange: with the shim, the handler
                 // re-enters inside it each round rather than needing a new one.
-                /** @var McpFiber $fiber */
-                $fiber = new \Fiber(static function () use ($handler, $request, $session, $shim, $codec): Response|Error {
+                $exchange = static function () use ($handler, $request, $session, $shim, $codec): Response|Error {
                     $result = $handler->handle($request, $session);
 
                     return $shim?->fulfill($result, $handler, $request, $session, $codec) ?? $result;
-                });
+                };
+
+                /** @var McpFiber $fiber */
+                $fiber = new \Fiber(static fn (): Response|Error => self::runShielded($exchange));
 
                 $result = $fiber->start();
 
@@ -356,6 +358,39 @@ private function handleRequest(TransportInterface $transport, Request $request,
         }
     }
 
+    /**
+     * Runs the exchange in a fiber of its own, so only MCP payloads leave it.
+     *
+     * Fibers are process-wide, so any library between here and the handler may
+     * suspend the running fiber for its own scheduling, with no protocol
+     * meaning. The SDK has no scheduler to hand such a suspension to, so it is
+     * resumed on the spot; a {@see FiberSuspend} is re-yielded to the session
+     * fiber, and the peer's answer handed back to the handler.
+     *
+     * @param callable(): (Response<array<string, mixed>>|Error) $exchange
+     *
+     * @return Response<array<string, mixed>>|Error
+     */
+    private static function runShielded(callable $exchange): Response|Error
+    {
+        /** @var \Fiber<null, mixed, Response<array<string, mixed>>|Error, mixed> $fiber */
+        $fiber = new \Fiber($exchange);
+        $yielded = $fiber->start();
+
+        while (!$fiber->isTerminated()) {
+            if (\is_array($yielded) && isset($yielded['type'])) {
+                /** @var FiberSuspend $yielded */
+                $yielded = $fiber->resume(\Fiber::suspend($yielded));
+
+                continue;
+            }
+
+            $yielded = $fiber->resume();
+        }
+
+        return $fiber->getReturn();
+    }
+
     /**
      * @param Response<array<string, mixed>>|Error $response
      */

tests/Unit/Server/ProtocolTest.php

@@ -795,6 +795,142 @@ public function testHandlerUnexpectedExceptionReturnsInternalError(): void
         $this->assertStringNotContainsString('Unexpected error', $message['error']['message']);
     }
 
+    #[TestDox('A suspension from the host framework is driven to completion in band')]
+    public function testForeignFiberSuspensionDoesNotReachTheTransport(): void
+    {
+        $handler = $this->createMock(RequestHandlerInterface::class);
+        $handler->method('supports')->willReturn(true);
+        $handler->method('handle')->willReturnCallback(static function (): Response {
+            // A host framework batching slow lookups suspends the running fiber
+            // with a value of its own; Drupal's entity loader passes a resume
+            // hint, its theme registry passes nothing at all.
+            \Fiber::suspend('Immediate');
+            \Fiber::suspend(null);
+
+            return new Response(1, ['status' => 'ok']);
+        });
+
+        $this->transport->expects($this->never())->method('attachFiberToSession');
+
+        $sessionManager = new SessionManager(new InMemorySessionStore());
+
+        $protocol = new Protocol(
+            requestHandlers: [$handler],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $sessionManager,
+        );
+
+        $session = $sessionManager->create();
+        $session->save();
+        $sessionId = $session->getId();
+
+        $protocol->processInput(
+            $this->transport,
+            '{"jsonrpc": "2.0", "id": 1, "method": "ping"}',
+            $sessionId
+        );
+
+        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        $this->assertCount(1, $outgoing);
+
+        $message = json_decode($outgoing[0]['message'], true);
+        $this->assertSame(['status' => 'ok'], $message['result']);
+    }
+
+    #[TestDox('An outbound notification still reaches the transport')]
+    public function testOutboundNotificationStillReachesTheTransport(): void
+    {
+        $handler = $this->createMock(RequestHandlerInterface::class);
+        $handler->method('supports')->willReturn(true);
+        $handler->method('handle')->willReturnCallback(static function (): Response {
+            \Fiber::suspend('Immediate');
+            \Fiber::suspend([
+                'type' => 'notification',
+                'notification' => new LoggingMessageNotification(LoggingLevel::Info, 'working'),
+            ]);
+
+            return new Response(1, ['status' => 'ok']);
+        });
+
+        $this->transport->expects($this->once())->method('attachFiberToSession');
+
+        $sessionManager = new SessionManager(new InMemorySessionStore());
+
+        $protocol = new Protocol(
+            requestHandlers: [$handler],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $sessionManager,
+        );
+
+        $session = $sessionManager->create();
+        $session->save();
+        $sessionId = $session->getId();
+
+        $protocol->processInput(
+            $this->transport,
+            '{"jsonrpc": "2.0", "id": 1, "method": "ping"}',
+            $sessionId
+        );
+
+        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        $this->assertCount(1, $outgoing);
+
+        $message = json_decode($outgoing[0]['message'], true);
+        $this->assertSame(LoggingMessageNotification::getMethod(), $message['method']);
+    }
+
+    #[TestDox('An outbound request still receives the peer answer, around foreign suspensions')]
+    public function testOutboundRequestStillReceivesThePeerAnswer(): void
+    {
+        $answered = null;
+
+        $handler = $this->createMock(RequestHandlerInterface::class);
+        $handler->method('supports')->willReturn(true);
+        $handler->method('handle')->willReturnCallback(static function () use (&$answered): Response {
+            \Fiber::suspend('Immediate');
+            $answered = \Fiber::suspend(['type' => 'request', 'request' => new PingRequest(), 'timeout' => 5]);
+            \Fiber::suspend('Immediate');
+
+            return new Response(1, ['status' => 'ok']);
+        });
+
+        $sessionFiber = null;
+        $this->transport->method('attachFiberToSession')->willReturnCallback(
+            static function (\Fiber $fiber) use (&$sessionFiber): void {
+                $sessionFiber = $fiber;
+            }
+        );
+
+        $sessionManager = new SessionManager(new InMemorySessionStore());
+
+        $protocol = new Protocol(
+            requestHandlers: [$handler],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $sessionManager,
+        );
+
+        $session = $sessionManager->create();
+        $session->save();
+
+        $protocol->processInput(
+            $this->transport,
+            '{"jsonrpc": "2.0", "id": 1, "method": "ping"}',
+            $session->getId()
+        );
+
+        $this->assertInstanceOf(\Fiber::class, $sessionFiber);
+
+        $peerAnswer = new Response(2, ['pong' => true]);
+        $sessionFiber->resume($peerAnswer);
+
+        $this->assertTrue($sessionFiber->isTerminated());
+        $this->assertSame($peerAnswer, $answered);
+        $this->assertEquals(new Response(1, ['status' => 'ok']), $sessionFiber->getReturn());
+    }
+
     #[TestDox('Failure while dispatching an outbound request is answered under the inbound request id')]
     public function testOutboundRequestFailureIsAnsweredUnderInboundRequestId(): void
     {