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 {