From 21429a5e098a7a1a2e4aaf5e5d327c6186a4b5e8 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 7 Oct 2026 20:09:29 +0000
Subject: [PATCH] [Server] Stop overwriting client responses while polling for
 them

While a request to the client is pending over HTTP, the process holding
the SSE stream polls the session every 100ms, and every poll emptied the
outgoing queue and saved the whole session, even with nothing queued. The
client's answer arrives as a separate POST, usually in another process,
which stores it in the same session. A poll that read the session just
before that store and saved just after wrote the old copy back, the answer
was lost, and the request timed out.

Leave the session untouched when there is nothing to dequeue, so the
waiting loop only reads until the answer is there. This showed up as
intermittent timeouts of the multi-worker php -S tests in CI.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RjzbQfz2DVm9xHmGsqM7d
---
 CHANGELOG.md                                  |   1 +
 src/Server/Protocol.php                       |   5 +
 tests/Unit/Server/ProtocolSessionRaceTest.php | 110 ++++++++++++++++++
 3 files changed, 116 insertions(+)
 create mode 100644 tests/Unit/Server/ProtocolSessionRaceTest.php

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 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Tests\Unit\Server;
+
+use Mcp\JsonRpc\MessageFactory;
+use Mcp\Schema\JsonRpc\Response;
+use Mcp\Server\Protocol;
+use Mcp\Server\Session\InMemorySessionStore;
+use Mcp\Server\Session\SessionManager;
+use Mcp\Server\Session\SessionStoreInterface;
+use Mcp\Server\Transport\TransportInterface;
+use PHPUnit\Framework\Attributes\TestDox;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Uid\Uuid;
+
+/**
+ * Two workers sharing one session: one holds a tool call's SSE stream open and
+ * polls the session for the client's answer, the other receives that answer as
+ * a separate POST and stores it.
+ */
+final class ProtocolSessionRaceTest extends TestCase
+{
+    #[TestDox('polling the session does not overwrite a client response stored meanwhile')]
+    public function testPollingDoesNotOverwriteAConcurrentResponse(): void
+    {
+        $store = new InterleavingSessionStore(new InMemorySessionStore());
+        $sessions = new SessionManager($store, gcProbability: 0);
+        $sessionId = Uuid::v4();
+        $sessions->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();
+    }
+}