diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d9a859a..c61e6266 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * [BC Break] `AbstractSchemaDefinition` declares an abstract `getDefault()`, which a custom schema definition has to implement. * Fix `Client::getPrompt()` without arguments sending `"arguments": []`, which servers validating the spec's object type (e.g. the TypeScript SDK) reject: empty arguments are now omitted. * Add `Tool::$execution` (`ToolExecution` with a `TaskSupport` enum) and `ServerCapabilities::$tasks`, which were dropped when parsing a 2025-11-25 server's `tools/list` and `initialize` results. +* Fix a request to the client (sampling, elicitation, roots) intermittently timing out over HTTP when the server runs in several processes: the stream waiting for the answer saved the session on every poll, and could overwrite the answer another process had just stored. 0.8.0 ----- diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index e92991f3..a441542d 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -518,6 +518,11 @@ public function consumeOutgoingMessages(Uuid $sessionId): array { $session = $this->sessionManager->createWithId($sessionId); $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []); + + if ([] === $queue) { + return []; + } + $session->set(self::SESSION_OUTGOING_QUEUE, []); $session->save(); diff --git a/tests/Unit/Server/ProtocolSessionRaceTest.php b/tests/Unit/Server/ProtocolSessionRaceTest.php new file mode 100644 index 00000000..da122ccd --- /dev/null +++ b/tests/Unit/Server/ProtocolSessionRaceTest.php @@ -0,0 +1,110 @@ +createWithId($sessionId)->save(); + + $waiting = new Protocol([], [], MessageFactory::make(), $sessions); + $answering = new Protocol([], [], MessageFactory::make(), $sessions); + $transport = $this->createMock(TransportInterface::class); + + // The answer lands right after the waiting worker read the session, + // before anything it does next could write the session back. + $store->afterNextRead(static function () use ($answering, $transport, $sessionId): void { + $answering->processInput($transport, '{"jsonrpc": "2.0", "id": 7, "result": {"ok": true}}', $sessionId); + }); + + // One turn of the waiting worker's loop, with nothing queued to send. + $this->assertSame([], $waiting->consumeOutgoingMessages($sessionId)); + + $this->assertInstanceOf(Response::class, $waiting->checkResponse(7, $sessionId)); + } +} + +/** + * Runs a callback once, right after the next read, to interleave another + * worker's write with the reader's read-modify-write. + */ +final class InterleavingSessionStore implements SessionStoreInterface +{ + /** @var (\Closure(): void)|null */ + private ?\Closure $afterNextRead = null; + + public function __construct( + private readonly SessionStoreInterface $store, + ) { + } + + /** + * @param \Closure(): void $callback + */ + public function afterNextRead(\Closure $callback): void + { + $this->afterNextRead = $callback; + } + + public function exists(Uuid $id): bool + { + return $this->store->exists($id); + } + + public function read(Uuid $id): string|false + { + $data = $this->store->read($id); + + if (null !== $callback = $this->afterNextRead) { + $this->afterNextRead = null; + $callback(); + } + + return $data; + } + + public function write(Uuid $id, string $data): bool + { + return $this->store->write($id, $data); + } + + public function destroy(Uuid $id): bool + { + return $this->store->destroy($id); + } + + public function gc(): array + { + return $this->store->gc(); + } +}