From cf1a14f3d3e11292efb1eb8e41d3bcceea78992d Mon Sep 17 00:00:00 2001 From: loki et fago Date: Thu, 10 Sep 2026 21:03:04 +0200 Subject: [PATCH] [Server] Shield a request handler from foreign fiber suspends. Fibers are process-wide, so a host framework may suspend the running fiber between Protocol and the handler for its own scheduling. Protocol read every suspension as an MCP yield, stranding the handler with no message attached. Drive the handler from an inner fiber so only ClientGateway payloads reach the transport. --- src/Server/Protocol.php | 41 ++++++++- tests/Unit/Server/ProtocolTest.php | 136 +++++++++++++++++++++++++++++ 2 files changed, 174 insertions(+), 3 deletions(-) diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index f460c3c7..e5b05249 100644 --- a/src/Server/Protocol.php +++ b/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>|Error) $exchange + * + * @return Response>|Error + */ + private static function runShielded(callable $exchange): Response|Error + { + /** @var \Fiber>|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>|Error $response */ diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index 175db74f..befd1412 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/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 {