diff --git a/CHANGELOG.md b/CHANGELOG.md
index 91cf7edd1..7392d5d75 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -40,6 +40,8 @@ All notable changes to `mcp/sdk` will be documented in this file.
   * [BC Break] A handler calling `\Fiber::suspend()` itself with anything but a `NotificationSuspension` or `RequestSuspension` fails the request without a scheduler, also when no event loop is involved. It used to be resumed with `null`.
   * Fix a request failing with `503`, and a resumed request losing its session writes, while another request of the session holds the session lock across an await: both wait for the lock on the event loop.
   * Fix a handler still running after its request was answered, e.g. without a fiber scheduler: it no longer fires `ResponseEvent` or `ErrorEvent`, what it throws is logged, and calling the `ClientGateway` throws.
+* [BC Break] Bind handshake-era sessions to the principal that opened them: `AccessToken::getPrincipal()`, or the `StreamableHttpTransport::PRINCIPAL_ATTRIBUTE` request attribute when set. Requests of another principal, and to sessions opened before the upgrade, are answered with `404`, and to a session the store cannot read with `503`. `Protocol::processInput()`, the transports' `handleMessage()`, `handleSessionEnd()`, `handlePostRequest()` and `handleDeleteRequest()`, and the `onMessage()` and `onSessionEnd()` listeners take a `?string $principal` as last argument.
+* [BC Break] Answer malformed JSON and an `initialize` sent with a session id with HTTP `400`, and a failure before the session is checked with `500`, instead of `200`. Such an answer no longer carries, or removes, the messages queued for the session in `Mcp-Session-Id`.
 
 ### Client
 
diff --git a/docs/run/authorization.md b/docs/run/authorization.md
index 1375560ef..ce15e7d20 100644
--- a/docs/run/authorization.md
+++ b/docs/run/authorization.md
@@ -313,11 +313,27 @@ challenge a client steps up from. Use `$token->hasScope()` for decisions that de
 arguments beyond the tool name. With a `ScopePolicy`, the token's scopes include those implied by
 its hierarchy.
 
+## Sessions and the Principal
+
+A handshake-era session only serves the principal that opened it, as the
+[security best practices](https://modelcontextprotocol.io/specification/2025-11-25/basic/security_best_practices#session-hijacking)
+recommend. The principal is the issuer, client and subject of the access token. They stay the same
+when the client refreshes its token, so the session survives the refresh.
+
+When another principal sends the session's `Mcp-Session-Id`, the server answers 404, as for an
+unknown session, and a `DELETE` does not end the session.
+
+A token without a `sub` claim only identifies its issuer and client, so all tokens of one client share
+their sessions. If your authorization server identifies users by another claim, set the
+`StreamableHttpTransport::PRINCIPAL_ATTRIBUTE` request attribute from it. A session opened without a
+token only serves requests without a token.
+
 ## Authenticating Outside the SDK
 
 If your application or framework already authenticates the request, skip `AuthorizationMiddleware`
 and hand the result to the transport as the PSR-7 request attribute `AccessToken::class`. Both HTTP
-transports read it from there, so handlers get it through `RequestContext::getAccessToken()` as usual:
+transports read it from there, so handlers get it through `RequestContext::getAccessToken()` as usual,
+and a session only serves the principal of the token that opened it:
 
 ```php
 use Mcp\Server\Authorization\AccessToken;
@@ -326,6 +342,16 @@ $request = $request->withAttribute(AccessToken::class, new AccessToken($scopes,
 $transport = new StreamableHttpTransport($request);
 ```
 
+If the claims do not identify the user, or there is no token, set the
+`StreamableHttpTransport::PRINCIPAL_ATTRIBUTE` attribute as well. It takes a non-empty string, so cast
+an integer user id. The server answers any other value with a 500:
+
+```php
+use Mcp\Server\Transport\StreamableHttpTransport;
+
+$request = $request->withAttribute(StreamableHttpTransport::PRINCIPAL_ATTRIBUTE, (string) $userId);
+```
+
 ## Custom Token Validators
 
 Implement `AuthorizationTokenValidatorInterface` for other token formats, e.g. opaque tokens
diff --git a/src/Server/Authorization/AccessToken.php b/src/Server/Authorization/AccessToken.php
index 65b3e181e..0b3c39dc6 100644
--- a/src/Server/Authorization/AccessToken.php
+++ b/src/Server/Authorization/AccessToken.php
@@ -81,6 +81,16 @@ public function getClientId(): ?string
         return \is_string($clientId) ? $clientId : null;
     }
 
+    /**
+     * Who the token identifies: its issuer, client and subject, which outlive a token refresh.
+     */
+    public function getPrincipal(): string
+    {
+        $issuer = $this->getClaim('iss');
+
+        return serialize([\is_string($issuer) ? $issuer : null, $this->getClientId(), $this->getSubject()]);
+    }
+
     /**
      * Never serialized: the token must not outlive its request in a session store.
      */
diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php
index 3c4978878..b801bd609 100644
--- a/src/Server/Protocol.php
+++ b/src/Server/Protocol.php
@@ -74,6 +74,11 @@ class Protocol
 
     public const SESSION_LOGGING_LEVEL = '_mcp.logging_level';
 
+    private const SESSION_NOT_FOUND_MESSAGE = 'Session not found or has expired.';
+
+    /** Session key for the fingerprint of the principal that opened the session, null when it was opened without one */
+    private const SESSION_PRINCIPAL = '_mcp.principal';
+
     /**
      * Deliberately generic: unexpected throwables carry internal details such as file paths, class
      * names and argument types, which must not be handed to the peer. The full exception is logged.
@@ -147,7 +152,7 @@ public function connect(TransportInterface $transport): void
     {
         $transport->onMessage($this->processInput(...));
 
-        $transport->onSessionEnd($this->destroySession(...));
+        $transport->onSessionEnd($this->endSession(...));
 
         // The transport keeps these callbacks, so they reference it weakly to not keep it alive.
         $transportRef = \WeakReference::create($transport);
@@ -175,13 +180,14 @@ public function connect(TransportInterface $transport): void
      *
      * @param TransportInterface<mixed> $transport
      * @param AccessToken|null          $accessToken the token the input was authorized with, if the transport authorizes
+     * @param string|null               $principal   who the input was authenticated as, if the transport authenticates
      */
-    public function processInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken = null): void
+    public function processInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken = null, ?string $principal = null): void
     {
         // Last line of defense: a malformed message must never escape as a PHP error and take the
         // server process down.
         try {
-            $this->doProcessInput($transport, $input, $sessionId, $accessToken);
+            $this->doProcessInput($transport, $input, $sessionId, $accessToken, $principal);
         } catch (\Throwable $e) {
             $this->logger->error(\sprintf('Uncaught exception while processing input: %s', $e->getMessage()), ['exception' => $e]);
 
@@ -192,7 +198,7 @@ public function processInput(TransportInterface $transport, string $input, ?Uuid
             }
 
             try {
-                $this->sendResponse($transport, Error::forInternalError(self::INTERNAL_ERROR_MESSAGE, $id), null);
+                $this->sendResponse($transport, Error::forInternalError(self::INTERNAL_ERROR_MESSAGE, $id), null, ['status_code' => 500]);
             } catch (\Throwable $e) {
                 $this->logger->error(\sprintf('Failed to send internal error response: %s', $e->getMessage()), ['exception' => $e]);
             }
@@ -226,7 +232,7 @@ private static function findResponseId(string $input): string|int|null
     /**
      * @param TransportInterface<mixed> $transport
      */
-    private function doProcessInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken): void
+    private function doProcessInput(TransportInterface $transport, string $input, ?Uuid $sessionId, ?AccessToken $accessToken, ?string $principal): void
     {
         $this->logger->info('Received message to process.');
         $this->logger->debug('Received message payload.', ['message' => $input]);
@@ -238,7 +244,8 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
         } catch (\JsonException $e) {
             $this->logger->warning('Failed to decode json message.', ['exception' => $e]);
             $error = Error::forParseError($e->getMessage());
-            $this->sendResponse($transport, $error, null);
+            // The status keeps a refusal from carrying the queue of a session nobody checked yet, here and below.
+            $this->sendResponse($transport, $error, null, ['status_code' => 400]);
 
             return;
         } catch (InvalidInputMessageException $e) {
@@ -247,13 +254,29 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
         }
 
         // Taken before the session is read and held until it is saved, so no other request of
-        // the session reads a state this one is about to replace. A session created here for
-        // an `initialize` is not known to anyone else yet and needs no lock, and an unknown
-        // session id is refused by resolveSession() without one; the session is looked up
-        // again under the lock, as it may have ended while this request waited.
-        $takesLock = null !== $sessionId && null !== $this->sessionLock && $this->sessionManager->exists($sessionId);
+        // the session reads a state this one is about to replace. An `initialize` never reads an
+        // existing session and an unknown session id is refused by resolveSession(), so neither
+        // needs the lock. The session is looked up again under the lock, as it may have ended
+        // while this request waited.
+        $takesLock = null !== $sessionId && null !== $this->sessionLock && !$message instanceof InitializeRequest && $this->sessionManager->exists($sessionId);
 
         if ($takesLock) {
+            // Who opened the session is checked before waiting for the lock too, so that another
+            // principal can neither tell the session is busy nor keep it busy.
+            try {
+                $serves = $this->servesPrincipal($this->sessionManager->createWithId($sessionId), $principal);
+            } catch (SessionStoreException $e) {
+                $this->refuseUnavailableStore($transport, $sessionId, self::findResponseId($input), $e);
+
+                return;
+            }
+
+            if (!$serves) {
+                $this->refuseOtherPrincipal($transport, $sessionId);
+
+                return;
+            }
+
             try {
                 $refusal = $this->acquireSessionLock($sessionId) ? null : Error::forServerError(self::SESSION_BUSY_MESSAGE, self::findResponseId($input));
             } catch (SessionLockException) {
@@ -268,7 +291,7 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
         }
 
         try {
-            $this->handleMessage($transport, $sessionId, $message, $accessToken);
+            $this->handleMessage($transport, $sessionId, $message, $accessToken, $principal);
         } finally {
             if ($takesLock) {
                 $this->releaseSessionLock($sessionId);
@@ -279,9 +302,9 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
     /**
      * @param TransportInterface<mixed> $transport
      */
-    private function handleMessage(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message, ?AccessToken $accessToken): void
+    private function handleMessage(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message, ?AccessToken $accessToken, ?string $principal): void
     {
-        $session = $this->resolveSession($transport, $sessionId, $message);
+        $session = $this->resolveSession($transport, $sessionId, $message, $principal);
         if (null === $session) {
             return;
         }
@@ -1107,19 +1130,22 @@ private function errorForThrowable(\Throwable $e, Request $request, SessionInter
      *
      * @param TransportInterface<mixed> $transport
      * @param Uuid|null                 $sessionId The session ID from the transport
+     * @param string|null               $principal Who the messages were authenticated as
      */
-    private function resolveSession(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message): ?SessionInterface
+    private function resolveSession(TransportInterface $transport, ?Uuid $sessionId, MessageInterface|InvalidInputMessageException $message, ?string $principal): ?SessionInterface
     {
         if ($message instanceof InitializeRequest) {
             // Spec: An initialize request must not have a session ID.
             if ($sessionId) {
                 $error = Error::forInvalidRequest('A session ID MUST NOT be sent with an "initialize" request.');
-                $this->sendResponse($transport, $error, null);
+                $this->sendResponse($transport, $error, null, ['status_code' => 400]);
 
                 return null;
             }
 
             $session = $this->sessionManager->create();
+            $session->set(self::SESSION_PRINCIPAL, self::fingerprint($principal));
+
             $this->logger->debug('Created new session for initialize', [
                 'session_id' => $session->getId()->toRfc4122(),
             ]);
@@ -1140,25 +1166,88 @@ private function resolveSession(TransportInterface $transport, ?Uuid $sessionId,
         try {
             $exists = $this->sessionManager->exists($sessionId);
         } catch (SessionStoreException $e) {
-            // Not a 404: the session may well exist, so the client should retry instead of re-initializing.
-            $this->logger->error('Session store is unavailable.', [
-                'session_id' => $sessionId->toRfc4122(),
-                'exception' => $e,
-            ]);
-            $error = Error::forInternalError('Session store is unavailable.', self::findRequestId($message));
-            $this->sendResponse($transport, $error, null, ['status_code' => 503]);
+            $this->refuseUnavailableStore($transport, $sessionId, self::findRequestId($message), $e);
 
             return null;
         }
 
         if (!$exists) {
-            $error = Error::forInvalidRequest('Session not found or has expired.');
+            $error = Error::forInvalidRequest(self::SESSION_NOT_FOUND_MESSAGE);
             $this->sendResponse($transport, $error, null, ['status_code' => 404]);
 
             return null;
         }
 
-        return $this->sessionManager->createWithId($sessionId);
+        $session = $this->sessionManager->createWithId($sessionId);
+
+        try {
+            $serves = $this->servesPrincipal($session, $principal);
+        } catch (SessionStoreException $e) {
+            $this->refuseUnavailableStore($transport, $sessionId, self::findRequestId($message), $e);
+
+            return null;
+        }
+
+        if (!$serves) {
+            $this->refuseOtherPrincipal($transport, $sessionId);
+
+            return null;
+        }
+
+        return $session;
+    }
+
+    /**
+     * Whether the session serves the principal: the one that opened it, or no principal for a session opened without one.
+     *
+     * A session that does not record who opened it serves no one.
+     *
+     * @throws SessionStoreException when the session exists but its data cannot be read
+     */
+    private function servesPrincipal(SessionInterface $session, ?string $principal): bool
+    {
+        if ($session->get(self::SESSION_PRINCIPAL, false) === self::fingerprint($principal)) {
+            return true;
+        }
+
+        if ([] === $session->all()) {
+            throw new SessionStoreException(\sprintf('Session "%s" could not be read.', $session->getId()->toRfc4122()));
+        }
+
+        return false;
+    }
+
+    /**
+     * Not a 404: the session may well exist, so the client should retry instead of re-initializing.
+     *
+     * @param TransportInterface<mixed> $transport
+     */
+    private function refuseUnavailableStore(TransportInterface $transport, Uuid $sessionId, string|int|null $requestId, SessionStoreException $exception): void
+    {
+        $this->logger->error('Session store is unavailable.', [
+            'session_id' => $sessionId->toRfc4122(),
+            'exception' => $exception,
+        ]);
+        $this->sendResponse($transport, Error::forInternalError('Session store is unavailable.', $requestId), null, ['status_code' => 503]);
+    }
+
+    /**
+     * @param TransportInterface<mixed> $transport
+     */
+    private function refuseOtherPrincipal(TransportInterface $transport, Uuid $sessionId): void
+    {
+        $this->logger->warning('Refused a request of another principal than the one that opened the session.', [
+            'session_id' => $sessionId->toRfc4122(),
+        ]);
+
+        // Answered like an unknown session, so that another principal cannot tell the session exists.
+        $error = Error::forInvalidRequest(self::SESSION_NOT_FOUND_MESSAGE);
+        $this->sendResponse($transport, $error, null, ['status_code' => 404]);
+    }
+
+    private static function fingerprint(?string $principal): ?string
+    {
+        return null === $principal ? null : hash('sha256', $principal);
     }
 
     /**
@@ -1173,6 +1262,26 @@ private static function findRequestId(MessageInterface|InvalidInputMessageExcept
         };
     }
 
+    /**
+     * Ends a session at the request of a client, which must be the principal that opened it.
+     *
+     * @return bool false when the session does not exist, or is another principal's
+     *
+     * @throws SessionStoreException when who opened the session cannot be established; the session is not ended
+     */
+    private function endSession(Uuid $sessionId, ?string $principal = null): bool
+    {
+        if ($this->sessionManager->exists($sessionId) && !$this->servesPrincipal($this->sessionManager->createWithId($sessionId), $principal)) {
+            $this->logger->warning('Refused to end a session for another principal than the one that opened it.', [
+                'session_id' => $sessionId->toRfc4122(),
+            ]);
+
+            return false;
+        }
+
+        return $this->destroySession($sessionId);
+    }
+
     /**
      * Destroy a specific session.
      *
diff --git a/src/Server/Transport/BaseTransport.php b/src/Server/Transport/BaseTransport.php
index 451915b6f..7b3da98df 100644
--- a/src/Server/Transport/BaseTransport.php
+++ b/src/Server/Transport/BaseTransport.php
@@ -11,6 +11,7 @@
 
 namespace Mcp\Server\Transport;
 
+use Mcp\Exception\SessionStoreException;
 use Mcp\Schema\JsonRpc\Error;
 use Mcp\Schema\JsonRpc\Response;
 use Mcp\Server\Authorization\AccessToken;
@@ -166,20 +167,34 @@ protected function resumeHandlerFiber(\Fiber $fiber, ?Uuid $sessionId, Response|
         }
     }
 
-    protected function handleMessage(string $payload, ?Uuid $sessionId, ?AccessToken $accessToken = null): void
+    protected function handleMessage(string $payload, ?Uuid $sessionId, ?AccessToken $accessToken = null, ?string $principal = null): void
     {
         if (\is_callable($this->messageListener)) {
-            ($this->messageListener)($this, $payload, $sessionId, $accessToken);
+            ($this->messageListener)($this, $payload, $sessionId, $accessToken, $principal);
+        }
+    }
+
+    /**
+     * Ends the session of a transport shutting down, which has no client to report a store failure to.
+     */
+    protected function endSessionOnShutdown(?Uuid $sessionId): void
+    {
+        try {
+            $this->handleSessionEnd($sessionId);
+        } catch (SessionStoreException $e) {
+            $this->logger->error('Session store is unavailable, the session was not ended.', ['exception' => $e]);
         }
     }
 
     /**
      * @return bool false when the listener reported that the session does not exist
+     *
+     * @throws SessionStoreException when the listener cannot tell who opened the session
      */
-    protected function handleSessionEnd(?Uuid $sessionId): bool
+    protected function handleSessionEnd(?Uuid $sessionId, ?string $principal = null): bool
     {
         if ($sessionId && \is_callable($this->sessionEndListener)) {
-            return false !== ($this->sessionEndListener)($sessionId);
+            return false !== ($this->sessionEndListener)($sessionId, $principal);
         }
 
         return true;
diff --git a/src/Server/Transport/InMemoryTransport.php b/src/Server/Transport/InMemoryTransport.php
index ddc6ca122..c5021fe61 100644
--- a/src/Server/Transport/InMemoryTransport.php
+++ b/src/Server/Transport/InMemoryTransport.php
@@ -55,7 +55,7 @@ public function listen(): mixed
         }
 
         $this->logger->info('InMemoryTransport finished processing.');
-        $this->handleSessionEnd($this->sessionId);
+        $this->endSessionOnShutdown($this->sessionId);
 
         $this->sessionId = null;
 
@@ -69,7 +69,7 @@ public function setSessionId(?Uuid $sessionId): void
 
     public function close(): void
     {
-        $this->handleSessionEnd($this->sessionId);
+        $this->endSessionOnShutdown($this->sessionId);
         $this->sessionId = null;
     }
 }
diff --git a/src/Server/Transport/ManagesTransportCallbacks.php b/src/Server/Transport/ManagesTransportCallbacks.php
index 8ca342e91..d315cf9dd 100644
--- a/src/Server/Transport/ManagesTransportCallbacks.php
+++ b/src/Server/Transport/ManagesTransportCallbacks.php
@@ -28,10 +28,10 @@
  * */
 trait ManagesTransportCallbacks
 {
-    /** @var callable(TransportInterface<mixed>, string, ?Uuid, ?AccessToken): void */
+    /** @var callable(TransportInterface<mixed>, string, ?Uuid, ?AccessToken, ?string): void */
     protected $messageListener;
 
-    /** @var callable(Uuid): bool */
+    /** @var callable(Uuid, ?string): bool */
     protected $sessionEndListener;
 
     /** @var callable(Uuid): array<int, array{message: string, context: array<string, mixed>}> */
diff --git a/src/Server/Transport/StdioTransport.php b/src/Server/Transport/StdioTransport.php
index 55ffaeaf5..e4b2fc9e0 100644
--- a/src/Server/Transport/StdioTransport.php
+++ b/src/Server/Transport/StdioTransport.php
@@ -155,7 +155,7 @@ public function listen(): int
         $this->logger->info('StdioTransport finished listening.');
         if (\in_array($this->runnerControl->getState(), [RunnerState::RUNNING, RunnerState::STOP_AND_END_SESSION], true)) {
             $this->logger->info('StdioTransport end session.');
-            $this->handleSessionEnd($this->sessionId);
+            $this->endSessionOnShutdown($this->sessionId);
         }
 
         return 0;
@@ -459,7 +459,7 @@ private function writeLine(string $payload): void
 
     public function close(): void
     {
-        $this->handleSessionEnd($this->sessionId);
+        $this->endSessionOnShutdown($this->sessionId);
         if (\is_resource($this->input)) {
             fclose($this->input);
         }
diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php
index 6a57c4143..8d1421a38 100644
--- a/src/Server/Transport/StreamableHttpTransport.php
+++ b/src/Server/Transport/StreamableHttpTransport.php
@@ -13,6 +13,7 @@
 
 use Http\Discovery\Psr17FactoryDiscovery;
 use Mcp\Exception\InvalidArgumentException;
+use Mcp\Exception\SessionStoreException;
 use Mcp\Schema\Enum\ProtocolVersion;
 use Mcp\Schema\JsonRpc\Error;
 use Mcp\Server\Authorization\AccessToken;
@@ -59,6 +60,14 @@ class StreamableHttpTransport extends BaseTransport implements StatelessAwareTra
     public const SESSION_HEADER = 'Mcp-Session-Id';
     public const PROTOCOL_VERSION_HEADER = 'Mcp-Protocol-Version';
 
+    /**
+     * PSR-7 request attribute holding a non-empty string that identifies the authenticated principal.
+     *
+     * A session only serves the principal that opened it. Without this attribute, the principal is
+     * the one of the request's {@see AccessToken}; set it to identify users another way.
+     */
+    public const PRINCIPAL_ATTRIBUTE = 'mcp.principal';
+
     /**
      * Upper bound on the request body read for a POST, guarding against memory
      * exhaustion from an oversized (or unbounded chunked) payload.
@@ -206,10 +215,11 @@ protected function handleOptionsRequest(): ResponseInterface
     /**
      * @param string           $body        the request body, already read and bounded by {@see self::handleRequest()}
      * @param AccessToken|null $accessToken the token the request was authorized with, see {@see self::accessToken()}
+     * @param string|null      $principal   the principal the request was authenticated as, see {@see self::PRINCIPAL_ATTRIBUTE}
      */
-    protected function handlePostRequest(string $body, ?AccessToken $accessToken = null): ResponseInterface
+    protected function handlePostRequest(string $body, ?AccessToken $accessToken = null, ?string $principal = null): ResponseInterface
     {
-        $this->handleMessage($body, $this->sessionId, $accessToken);
+        $this->handleMessage($body, $this->sessionId, $accessToken, $principal);
 
         // Consume the immediate response exactly once, so a transport instance
         // reused for a later POST does not replay it.
@@ -235,13 +245,24 @@ protected function handlePostRequest(string $body, ?AccessToken $accessToken = n
         return $this->createJsonResponse();
     }
 
-    protected function handleDeleteRequest(): ResponseInterface
+    /**
+     * @param string|null $principal the principal the request was authenticated as, see {@see self::PRINCIPAL_ATTRIBUTE}
+     */
+    protected function handleDeleteRequest(?string $principal = null): ResponseInterface
     {
         if (!$this->sessionId) {
             return $this->createErrorResponse(Error::forInvalidRequest(self::SESSION_HEADER.' header is required.'), 400);
         }
 
-        if (!$this->handleSessionEnd($this->sessionId)) {
+        try {
+            $ended = $this->handleSessionEnd($this->sessionId, $principal);
+        } catch (SessionStoreException $e) {
+            $this->logger->error('Session store is unavailable.', ['session_id' => $this->sessionId->toRfc4122(), 'exception' => $e]);
+
+            return $this->createErrorResponse(Error::forInternalError('Session store is unavailable.'), 503);
+        }
+
+        if (!$ended) {
             return $this->createErrorResponse(Error::forInvalidRequest('Session not found or has expired.'), 404);
         }
 
@@ -517,9 +538,17 @@ private function handleHandshakeRequest(ServerRequestInterface $request, ?string
             return $this->createErrorResponse(Error::forInvalidRequest(self::SESSION_HEADER.' header must be a valid UUID.'), 400);
         }
 
+        $principal = $request->getAttribute(self::PRINCIPAL_ATTRIBUTE) ?? self::accessToken($request)?->getPrincipal();
+
+        if (null !== $principal && (!\is_string($principal) || '' === $principal)) {
+            $this->logger->error(\sprintf('The "%s" request attribute must be a non-empty string, got "%s".', self::PRINCIPAL_ATTRIBUTE, get_debug_type($principal)));
+
+            return $this->createErrorResponse(Error::forInternalError('Internal error'), 500);
+        }
+
         return match ($request->getMethod()) {
-            'POST' => $this->handlePostRequest($body ?? '', self::accessToken($request)),
-            'DELETE' => $this->handleDeleteRequest(),
+            'POST' => $this->handlePostRequest($body ?? '', self::accessToken($request), $principal),
+            'DELETE' => $this->handleDeleteRequest($principal),
             default => $this->createErrorResponse(Error::forInvalidRequest('Method Not Allowed'), 405),
         };
     }
diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php
index a86bbfb44..3f1f231b4 100644
--- a/src/Server/Transport/TransportInterface.php
+++ b/src/Server/Transport/TransportInterface.php
@@ -72,8 +72,9 @@ public function close(): void;
      * The transport calls this whenever ANY message arrives, regardless of source.
      *
      * The access token is the one the message was authorized with, if the transport authorizes requests.
+     * The principal identifies who the message was authenticated as, if the transport authenticates requests.
      *
-     * @param callable(TransportInterface<TResult> $transport, string $message, ?Uuid $sessionId, ?AccessToken $accessToken): void $listener
+     * @param callable(TransportInterface<TResult> $transport, string $message, ?Uuid $sessionId, ?AccessToken $accessToken, ?string $principal): void $listener
      */
     public function onMessage(callable $listener): void;
 
@@ -81,8 +82,10 @@ public function onMessage(callable $listener): void;
      * Register a listener for when a session is terminated.
      *
      * The transport calls this when a client disconnects or explicitly ends their session.
+     * The principal identifies who ended it, if the transport authenticates requests.
+     * The listener throws a {@see \Mcp\Exception\SessionStoreException} when it cannot tell who opened the session.
      *
-     * @param callable(Uuid $sessionId): bool $listener The callback function to execute when destroying a session, returning false when the session does not exist
+     * @param callable(Uuid $sessionId, ?string $principal): bool $listener The callback function to execute when destroying a session, returning false when the session does not exist
      */
     public function onSessionEnd(callable $listener): void;
 
diff --git a/tests/Unit/Server/Authorization/AccessTokenTest.php b/tests/Unit/Server/Authorization/AccessTokenTest.php
index fd9bc23ef..b4e94c495 100644
--- a/tests/Unit/Server/Authorization/AccessTokenTest.php
+++ b/tests/Unit/Server/Authorization/AccessTokenTest.php
@@ -34,6 +34,39 @@ public function testFallsBackToAuthorizedParty(): void
         $this->assertNull((new AccessToken([], ['sub' => 42]))->getSubject());
     }
 
+    public function testPrincipalOutlivesARefresh(): void
+    {
+        $principal = (new AccessToken(['read'], ['iss' => 'https://as', 'sub' => 'alice', 'azp' => 'ide', 'jti' => '1']))->getPrincipal();
+
+        $this->assertSame($principal, (new AccessToken(['read', 'write'], ['iss' => 'https://as', 'sub' => 'alice', 'azp' => 'ide', 'jti' => '2']))->getPrincipal());
+        $this->assertSame($principal, (new AccessToken([], ['iss' => 'https://as', 'sub' => 'alice', 'client_id' => 'ide']))->getPrincipal());
+    }
+
+    public function testPrincipalDiffersByIssuerClientAndSubject(): void
+    {
+        $principal = (new AccessToken([], ['iss' => 'https://as', 'sub' => 'alice', 'azp' => 'ide']))->getPrincipal();
+
+        $this->assertNotSame($principal, (new AccessToken([], ['iss' => 'https://other', 'sub' => 'alice', 'azp' => 'ide']))->getPrincipal());
+        $this->assertNotSame($principal, (new AccessToken([], ['iss' => 'https://as', 'sub' => 'alice', 'azp' => 'cli']))->getPrincipal());
+        $this->assertNotSame($principal, (new AccessToken([], ['iss' => 'https://as', 'sub' => 'carol', 'azp' => 'ide']))->getPrincipal());
+    }
+
+    public function testPrincipalKeepsClaimsApart(): void
+    {
+        $this->assertNotSame(
+            (new AccessToken([], ['client_id' => 'a', 'sub' => 'b']))->getPrincipal(),
+            (new AccessToken([], ['client_id' => 'b', 'sub' => 'a']))->getPrincipal(),
+        );
+    }
+
+    public function testPrincipalKeepsClaimsThatAreNotUtf8Apart(): void
+    {
+        $this->assertNotSame(
+            (new AccessToken([], ['sub' => "\xB1"]))->getPrincipal(),
+            (new AccessToken([], ['sub' => "\xB2"]))->getPrincipal(),
+        );
+    }
+
     public function testIsNeverSerialized(): void
     {
         $this->assertSame('{"token":null}', json_encode(['token' => new AccessToken(['a'], ['sub' => 'user'])]));
diff --git a/tests/Unit/Server/BuilderTest.php b/tests/Unit/Server/BuilderTest.php
index 2a704ca6c..e3c25fa85 100644
--- a/tests/Unit/Server/BuilderTest.php
+++ b/tests/Unit/Server/BuilderTest.php
@@ -249,20 +249,22 @@ public function testSetSessionLockIsWiredIntoTheProtocol(): void
             '{"jsonrpc":"2.0","id":2,"method":"ping"}',
         ]));
 
-        $this->assertCount(11, $log, implode(', ', $log));
+        $this->assertCount(13, $log, implode(', ', $log));
         $sessionId = substr($log[0], \strlen('read '));
         $this->assertSame([
             // initialize: the session is new, nobody else can know it yet
             "read $sessionId",
             "read $sessionId",
             "write $sessionId",
-            // ping: locked around read and save
+            // ping: who opened the session is checked before waiting for the lock, then locked around read and save
+            "read $sessionId",
             "acquire $sessionId",
             "read $sessionId",
             "read $sessionId",
             "write $sessionId",
             "release $sessionId",
-            // the transport ends the session when it is done
+            // the transport ends the session when it is done, which checks who opened it first
+            "read $sessionId",
             "acquire $sessionId",
             "destroy $sessionId",
             "release $sessionId",
diff --git a/tests/Unit/Server/ProtocolSessionLockTest.php b/tests/Unit/Server/ProtocolSessionLockTest.php
index ce66c00d3..fa2084c0b 100644
--- a/tests/Unit/Server/ProtocolSessionLockTest.php
+++ b/tests/Unit/Server/ProtocolSessionLockTest.php
@@ -31,6 +31,7 @@
 use Mcp\Tests\Unit\Server\Session\Fixture\RecordingLogger;
 use Mcp\Tests\Unit\Server\Session\Fixture\RecordingSessionLock;
 use Mcp\Tests\Unit\Server\Session\Fixture\RecordingSessionStore;
+use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\Attributes\TestDox;
 use PHPUnit\Framework\TestCase;
 use Psr\Log\LoggerInterface;
@@ -43,6 +44,8 @@
  */
 final class ProtocolSessionLockTest extends TestCase
 {
+    private const INITIALIZE = '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}';
+
     /** @var list<string> */
     private array $log = [];
 
@@ -53,7 +56,7 @@ protected function setUp(): void
         $this->store = new RecordingSessionStore($this->log);
     }
 
-    #[TestDox('a request holds the lock from before the session is read until after it is saved')]
+    #[TestDox('a request holds the lock from before the session is read for handling until after it is saved')]
     public function testRequestHoldsTheLockAroundReadAndSave(): void
     {
         $lock = new RecordingSessionLock($this->log);
@@ -64,6 +67,8 @@ public function testRequestHoldsTheLockAroundReadAndSave(): void
         $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId);
 
         $this->assertSame([
+            // only to check who opened the session
+            "read $sessionId",
             "acquire $sessionId",
             "read $sessionId",
             // save() reads again to write back only the keys this request changed
@@ -122,7 +127,8 @@ public function testSessionEndedWhileWaitingForTheLockIsRefused(): void
         $lock = new RecordingSessionLock($this->log);
         $sessionManager = $this->createMock(SessionManagerInterface::class);
         $sessionManager->method('exists')->willReturnOnConsecutiveCalls(true, false);
-        $sessionManager->expects($this->never())->method('createWithId');
+        // Read before the lock to check who opened the session, and not again once it ended.
+        $sessionManager->expects($this->once())->method('createWithId');
         $protocol = new Protocol([new PingHandler()], [], MessageFactory::make(), $sessionManager, sessionLock: $lock);
         $sessionId = Uuid::v4();
         $transport = new RecordingTransport();
@@ -148,8 +154,8 @@ public function testRequestWithoutSessionIdTakesNoLock(): void
         $this->assertSame(400, $transport->sent[0]['context']['status_code']);
     }
 
-    #[TestDox('a request whose session is locked by another request is refused before the session is read')]
-    public function testBusySessionIsRefusedWithoutReadingIt(): void
+    #[TestDox('a request whose session is locked by another request is refused after reading only who opened the session')]
+    public function testBusySessionIsRefusedReadingOnlyWhoOpenedIt(): void
     {
         $lock = new RecordingSessionLock($this->log, acquirable: false);
         $protocol = $this->protocol($lock);
@@ -158,7 +164,7 @@ public function testBusySessionIsRefusedWithoutReadingIt(): void
 
         $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId);
 
-        $this->assertSame([], $this->log);
+        $this->assertSame(["read $sessionId"], $this->log);
         $this->assertCount(1, $transport->sent);
         $this->assertSame(503, $transport->sent[0]['context']['status_code']);
 
@@ -168,6 +174,34 @@ public function testBusySessionIsRefusedWithoutReadingIt(): void
         $this->assertSame('Another request of this session is still being processed.', $error['error']['message']);
     }
 
+    /**
+     * @return iterable<string, array{string}>
+     */
+    public static function provideInputsOfAnotherPrincipal(): iterable
+    {
+        yield 'a request' => ['{"jsonrpc":"2.0","id":1,"method":"ping"}'];
+        yield 'an initialize request' => [self::INITIALIZE];
+    }
+
+    #[DataProvider('provideInputsOfAnotherPrincipal')]
+    #[TestDox('another principal is answered as for an unknown session without waiting for the lock, so it cannot tell the session is busy')]
+    public function testAnotherPrincipalIsRefusedWithoutWaitingForTheLock(string $input): void
+    {
+        $lock = new RecordingSessionLock($this->log);
+        $protocol = $this->protocol($lock);
+        $opening = new RecordingTransport();
+        $protocol->processInput($opening, self::INITIALIZE, null, null, 'alice');
+        $sessionId = $opening->sent[0]['context']['session_id'];
+        $lock->acquirable = false;
+
+        $unknown = new RecordingTransport();
+        $protocol->processInput($unknown, $input, Uuid::v4(), null, 'carol');
+        $transport = new RecordingTransport();
+        $protocol->processInput($transport, $input, $sessionId, null, 'carol');
+
+        $this->assertSame($unknown->sent, $transport->sent);
+    }
+
     #[TestDox('a busy session is logged as a warning, not as an error')]
     public function testBusySessionIsLoggedAsAWarning(): void
     {
@@ -191,7 +225,7 @@ public function testUnavailableLockIsRefusedAndLoggedWithItsCause(): void
 
         $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId);
 
-        $this->assertSame([], $this->log);
+        $this->assertSame(["read $sessionId"], $this->log);
         $this->assertCount(1, $transport->sent);
         $this->assertSame(503, $transport->sent[0]['context']['status_code']);
 
@@ -585,6 +619,8 @@ private function protocol(RecordingSessionLock $lock, LoggerInterface $logger =
     private function existingSession(array $data = []): Uuid
     {
         $session = new Session($this->store, Uuid::v4());
+        // Opened without a principal
+        $session->set('_mcp.principal', null);
         foreach ($data as $key => $value) {
             $session->set($key, $value);
         }
diff --git a/tests/Unit/Server/ProtocolSessionRaceTest.php b/tests/Unit/Server/ProtocolSessionRaceTest.php
index f62136242..e13912f6a 100644
--- a/tests/Unit/Server/ProtocolSessionRaceTest.php
+++ b/tests/Unit/Server/ProtocolSessionRaceTest.php
@@ -36,6 +36,7 @@ public function testPollingDoesNotOverwriteAConcurrentResponse(): void
         $sessions = new SessionManager($store, gcProbability: 0);
         $sessionId = Uuid::v4();
         $session = $sessions->createWithId($sessionId);
+        $session->set('_mcp.principal', null);
         $session->set('_mcp.pending_requests', ['00000000-0000-4000-8000-000000000007' => ['request_id' => '00000000-0000-4000-8000-000000000007', 'timeout' => 120, 'timestamp' => time()]]);
         $session->save();
 
@@ -62,6 +63,7 @@ public function testPollingATimedOutRequestDoesNotOverwriteAConcurrentResponse()
         $sessions = new SessionManager($store, gcProbability: 0);
         $sessionId = Uuid::v4();
         $session = $sessions->createWithId($sessionId);
+        $session->set('_mcp.principal', null);
         $session->set('_mcp.pending_requests', [
             '00000000-0000-4000-8000-000000000007' => ['request_id' => '00000000-0000-4000-8000-000000000007', 'timeout' => 5, 'timestamp' => time() - 10],
             '00000000-0000-4000-8000-000000000008' => ['request_id' => '00000000-0000-4000-8000-000000000008', 'timeout' => 120, 'timestamp' => time()],
diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php
index a924f1762..4861368f6 100644
--- a/tests/Unit/Server/ProtocolTest.php
+++ b/tests/Unit/Server/ProtocolTest.php
@@ -535,7 +535,9 @@ public function write(Uuid $id, string $data): bool
         $store->throws = $throws;
         $sessionManager = new SessionManager($store, gcProbability: 0);
         $sessionId = Uuid::v4();
-        $sessionManager->createWithId($sessionId)->save();
+        $session = $sessionManager->createWithId($sessionId);
+        $session->set('_mcp.principal', null);
+        $session->save();
         $store->failWrites = true;
 
         // Resumed, the fiber would wait for a client response to a request the session never stored.
@@ -598,7 +600,9 @@ public function write(Uuid $id, string $data): bool
         $store->throws = $throws;
         $sessionManager = new SessionManager($store, gcProbability: 0);
         $sessionId = Uuid::v4();
-        $sessionManager->createWithId($sessionId)->save();
+        $session = $sessionManager->createWithId($sessionId);
+        $session->set('_mcp.principal', null);
+        $session->save();
         $store->failingWrites = 1;
 
         $protocol = new Protocol(
@@ -637,6 +641,7 @@ public function testSessionWritesAfterResumeAreSaved(): void
         $sessionManager = new SessionManager(new InMemorySessionStore(), gcProbability: 0);
         $sessionId = Uuid::v4();
         $session = $sessionManager->createWithId($sessionId);
+        $session->set('_mcp.principal', null);
         $session->set('dropped', true);
         $session->save();
 
@@ -692,7 +697,11 @@ public function testFailingNotificationListenerDoesNotProduceResponse(): void
 
         $queue = [];
         $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            return '_mcp.outgoing_queue' === $key ? $queue : $default;
+            return match ($key) {
+                '_mcp.outgoing_queue' => $queue,
+                '_mcp.principal' => null,
+                default => $default,
+            };
         });
         $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
             if ('_mcp.outgoing_queue' === $key) {
@@ -934,6 +943,7 @@ public function testOutboundRequestFailureIsAnsweredUnderInboundRequestId(): voi
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $sessionId = $session->getId();
         $protocol->processInput(
@@ -1057,6 +1067,7 @@ public function testConcurrentWorkersSendClientRequestsUnderDistinctIds(): void
         $store = new InterleavingSessionStore();
         $sessionManager = new SessionManager($store, gcProbability: 0);
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $sessionId = $session->getId();
 
@@ -1101,6 +1112,7 @@ public function testStreamStopsPollingTimedOutRequest(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $sessionId = $session->getId();
 
@@ -1143,6 +1155,7 @@ private function startTwoStreamsWaitingOnClient(): array
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $sessionId = $session->getId();
 
@@ -1236,7 +1249,9 @@ public function testResponseIsSentNotQueued(): void
 
         $sessions = new SessionManager(new InMemorySessionStore(), gcProbability: 0);
         $sessionId = Uuid::v4();
-        $sessions->createWithId($sessionId)->save();
+        $session = $sessions->createWithId($sessionId);
+        $session->set('_mcp.principal', null);
+        $session->save();
 
         $transport = new RecordingTransport();
 
@@ -1355,7 +1370,6 @@ public function testRequestEventIsDispatched(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1424,7 +1438,6 @@ public function testRequestEventModificationIsUsed(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1460,7 +1473,6 @@ public function testRequestEventWithNullDispatcher(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1504,7 +1516,6 @@ public function testResponseEventIsDispatched(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1561,7 +1572,6 @@ public function testResponseEventModificationIsUsed(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1611,7 +1621,6 @@ public function testErrorEventIsDispatchedForErrorResult(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1663,7 +1672,6 @@ public function testErrorEventIsDispatchedForInvalidArgument(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1714,7 +1722,6 @@ public function testErrorEventIsDispatchedForGenericException(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
@@ -1761,7 +1768,6 @@ public function testErrorEventIsDispatchedForMethodNotFound(): void
 
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
-        $session->method('get')->willReturn([]);
         $session->expects($this->once())->method('save');
         $session->expects($this->atLeastOnce())->method('set');
 
@@ -2123,6 +2129,7 @@ public function testLateResponseToTimedOutRequestIsDropped(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->set('_mcp.pending_requests', ['00000000-0000-4000-8000-000000001000' => ['request_id' => '00000000-0000-4000-8000-000000001000', 'timeout' => 5, 'timestamp' => time() - 10]]);
         $session->save();
 
@@ -2250,6 +2257,7 @@ public function testRequestAwaitedByLocalFiberIsKeptPastGracePeriod(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $sessionId = $session->getId();
 
@@ -2336,6 +2344,7 @@ public function testSuspendedFibersDispatchResultUnderTheirOwnRequest(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', $session->getId());
         $protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": 2, "method": "ping"}', $session->getId());
@@ -2384,6 +2393,7 @@ public function write(Uuid $id, string $data): bool
         };
 
         $session = (new SessionManager($store))->create();
+        $session->set('_mcp.principal', null);
         $session->save();
 
         $fibers = [];
@@ -2455,6 +2465,7 @@ public function testResumedFiberTurnsThrowingResponseListenerIntoInternalError()
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', $session->getId());
 
@@ -2507,6 +2518,7 @@ public function testResumedFiberSurvivesThrowingErrorListener(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', $session->getId());
 
@@ -2556,6 +2568,7 @@ public function testResumedFiberTurnsThrowingHandlerIntoInternalError(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', $session->getId());
 
@@ -2590,6 +2603,7 @@ public function testClientResponseEventIsNotDispatchedForIdLessError(): void
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->save();
         $protocol->processInput($this->transport, '{"jsonrpc": "2.0", "id": null, "error": {"code": -32700, "message": "Parse error"}}', $session->getId());
     }
@@ -2792,6 +2806,7 @@ private function createProtocolWithPendingRequests(array $pending): array
         );
 
         $session = $sessionManager->create();
+        $session->set('_mcp.principal', null);
         $session->set('_mcp.pending_requests', $pending);
         $session->save();
 
@@ -2809,9 +2824,11 @@ private function createSessionWithPendingRequests(array $requestIds): SessionInt
         }
 
         $session = $this->createMock(SessionInterface::class);
-        $session->method('get')->willReturnCallback(
-            static fn (string $key, mixed $default = null): mixed => '_mcp.pending_requests' === $key ? $pending : $default
-        );
+        $session->method('get')->willReturnCallback(static fn (string $key, mixed $default = null): mixed => match ($key) {
+            '_mcp.pending_requests' => $pending,
+            '_mcp.principal' => null,
+            default => $default,
+        });
 
         return $session;
     }
diff --git a/tests/Unit/Server/Transport/AccessTokenFlowTest.php b/tests/Unit/Server/Transport/AccessTokenFlowTest.php
index 784cf4767..fe092d764 100644
--- a/tests/Unit/Server/Transport/AccessTokenFlowTest.php
+++ b/tests/Unit/Server/Transport/AccessTokenFlowTest.php
@@ -66,10 +66,10 @@ public function testHandshakeEraHandlerSeesTheTokenOfItsRequest(): void
         $headers = ['Mcp-Session-Id' => $session, 'MCP-Protocol-Version' => ProtocolVersion::V2025_11_25->value];
 
         $this->assertSame('alice', $this->text($this->serve($server, $this->request($call(2), 'alice', $headers))));
-        $this->assertSame('bob', $this->text($this->serve($server, $this->request($call(3), 'bob', $headers))));
+        $this->assertSame('alice write', $this->text($this->serve($server, $this->request($call(3), 'alice.write', $headers))));
 
-        // Without the authorization middleware, nothing of the earlier requests is left in the session.
-        $this->assertSame('anonymous', $this->text($this->serve($server, $this->request($call(4), null, $headers), [])));
+        // Nothing of the earlier requests is left in the session.
+        $this->assertSame('alice', $this->text($this->serve($server, $this->request($call(4), 'alice', $headers))));
     }
 
     public function testModernEraHandlerSeesTheToken(): void
@@ -142,7 +142,9 @@ public function testBatchDoesNotGetPastTheScopesOfItsMessages(): void
 
     public static function whoami(RequestContext $context): string
     {
-        return $context->getAccessToken()?->getSubject() ?? 'anonymous';
+        $token = $context->getAccessToken();
+
+        return null === $token ? 'anonymous' : implode(' ', [$token->getSubject(), ...$token->getScopes()]);
     }
 
     private function server(): Server
@@ -166,7 +168,9 @@ private function authorization(?ScopePolicy $scopePolicy = null): AuthorizationM
         $validator = new class implements AuthorizationTokenValidatorInterface {
             public function validate(string $accessToken): AuthorizationResult
             {
-                return AuthorizationResult::allow(new AccessToken([], ['sub' => $accessToken]));
+                [$subject, $scope] = explode('.', $accessToken, 2) + [1 => null];
+
+                return AuthorizationResult::allow(new AccessToken(null === $scope ? [] : [$scope], ['sub' => $subject]));
             }
         };
 
diff --git a/tests/Unit/Server/Transport/SessionBindingTest.php b/tests/Unit/Server/Transport/SessionBindingTest.php
new file mode 100644
index 000000000..6e94cfc58
--- /dev/null
+++ b/tests/Unit/Server/Transport/SessionBindingTest.php
@@ -0,0 +1,510 @@
+<?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\Transport;
+
+use Mcp\Exception\SessionStoreException;
+use Mcp\JsonRpc\MessageFactory;
+use Mcp\Schema\Enum\ProtocolVersion;
+use Mcp\Schema\Notification\ResourceUpdatedNotification;
+use Mcp\Server;
+use Mcp\Server\Authorization\AccessToken;
+use Mcp\Server\Handler\Request\PingHandler;
+use Mcp\Server\Protocol;
+use Mcp\Server\RequestContext;
+use Mcp\Server\Session\InMemorySessionStore;
+use Mcp\Server\Session\Session;
+use Mcp\Server\Session\SessionManager;
+use Mcp\Server\Transport\Http\Middleware\AuthorizationMiddleware;
+use Mcp\Server\Transport\Http\Middleware\PassthroughMiddleware;
+use Mcp\Server\Transport\Http\OAuth\AuthorizationResult;
+use Mcp\Server\Transport\Http\OAuth\AuthorizationTokenValidatorInterface;
+use Mcp\Server\Transport\Http\OAuth\ProtectedResourceMetadata;
+use Mcp\Server\Transport\StreamableHttpTransport;
+use Mcp\Tests\Unit\Server\Session\Fixture\RecordingSessionLock;
+use Nyholm\Psr7\Factory\Psr17Factory;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\ServerRequestInterface;
+use Psr\Http\Server\MiddlewareInterface;
+use Psr\Http\Server\RequestHandlerInterface;
+use Symfony\Component\Uid\Uuid;
+
+/**
+ * A handshake-era session only serves the principal that opened it.
+ */
+final class SessionBindingTest extends TestCase
+{
+    private const PING = '{"jsonrpc":"2.0","id":2,"method":"ping"}';
+    private const QUEUED_URI = 'urn:queued-for-alice';
+
+    private Psr17Factory $factory;
+    private UnreadableSessionStore $store;
+    private Server $server;
+
+    protected function setUp(): void
+    {
+        $this->factory = new Psr17Factory();
+        $this->store = new UnreadableSessionStore();
+        $this->server = Server::builder()
+            ->setServerInfo('test', '1.0.0')
+            ->setSession($this->store)
+            ->addTool(self::whoami(...), name: 'whoami')
+            ->build();
+    }
+
+    public function testSessionServesTheSubjectThatOpenedIt(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+
+        $this->assertSame('alice', $this->text($this->call($session, 'alice.ide.1')));
+    }
+
+    public function testRefreshedTokenKeepsTheSession(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+
+        $this->assertSame('alice', $this->text($this->call($session, 'alice.ide.2')));
+    }
+
+    public function testAnotherSubjectIsAnsweredLikeAnUnknownSession(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+        $unknown = $this->call(Uuid::v4()->toRfc4122(), 'carol.ide.1');
+
+        $response = $this->call($session, 'carol.ide.1');
+
+        $this->assertSame(404, $unknown->getStatusCode());
+        $this->assertSame(404, $response->getStatusCode());
+        $this->assertSame((string) $unknown->getBody(), (string) $response->getBody());
+    }
+
+    public function testAnotherClientOfTheSameSubjectIsRefused(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+
+        $this->assertSame(404, $this->call($session, 'alice.cli.1')->getStatusCode());
+    }
+
+    public function testRequestWithoutPrincipalIsRefused(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+
+        $this->assertSame(404, $this->call($session, null, [new PassthroughMiddleware()])->getStatusCode());
+    }
+
+    public function testDeleteOfAnotherPrincipalKeepsTheSession(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+        $unknown = $this->serve($this->request('DELETE', '', 'carol.ide.1', Uuid::v4()->toRfc4122()));
+
+        $response = $this->serve($this->request('DELETE', '', 'carol.ide.1', $session));
+        $this->assertSame(404, $response->getStatusCode());
+        $this->assertSame($unknown->getStatusCode(), $response->getStatusCode());
+        $this->assertSame((string) $unknown->getBody(), (string) $response->getBody());
+        $this->assertSame('alice', $this->text($this->call($session, 'alice.ide.1')));
+
+        $this->serve($this->request('DELETE', '', 'alice.ide.1', $session));
+        $this->assertSame(404, $this->call($session, 'alice.ide.1')->getStatusCode());
+    }
+
+    public function testSessionThatCannotBeReadIsRetriedByEveryone(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+        $passthrough = [new PassthroughMiddleware()];
+
+        $this->store->unreadable = true;
+        $this->assertSame(503, $this->call($session, 'alice.ide.1')->getStatusCode());
+        $this->assertSame(503, $this->call($session, null, $passthrough)->getStatusCode());
+        $this->assertSame(503, $this->serve($this->request('DELETE', '', 'alice.ide.1', $session))->getStatusCode());
+        $this->assertSame(503, $this->serve($this->request('DELETE', '', null, $session), $passthrough)->getStatusCode());
+        $this->store->unreadable = false;
+
+        $this->assertSame('alice', $this->text($this->call($session, 'alice.ide.1')));
+    }
+
+    public function testSessionThatCannotBeReadIsRetriedWithASessionLock(): void
+    {
+        $log = [];
+        $server = Server::builder()
+            ->setServerInfo('test', '1.0.0')
+            ->setSession($this->store)
+            ->setSessionLock(new RecordingSessionLock($log))
+            ->build();
+        $session = $this->server($server, $this->request('POST', $this->initializeRequest(), 'alice.ide.1'))->getHeaderLine('Mcp-Session-Id');
+        $log = [];
+
+        $this->store->unreadable = true;
+        $this->assertSame(503, $this->server($server, $this->request('POST', $this->toolCall(), 'alice.ide.1', $session))->getStatusCode());
+        $this->assertSame([], $log);
+    }
+
+    public function testDeleteDuringAStoreOutageKeepsTheSession(): void
+    {
+        $session = $this->initialize('alice.ide.1');
+
+        $this->store->unavailable = true;
+        $this->assertSame(503, $this->serve($this->request('DELETE', '', 'alice.ide.1', $session))->getStatusCode());
+        $this->store->unavailable = false;
+
+        $this->assertSame('alice', $this->text($this->call($session, 'alice.ide.1')));
+    }
+
+    public function testSessionFromBeforeTheUpgradeIsNotFound(): void
+    {
+        $session = Uuid::v4();
+        $this->store->write($session, json_encode(['initialized' => true], \JSON_THROW_ON_ERROR));
+
+        $this->assertSame(404, $this->call($session->toRfc4122(), 'alice.ide.1')->getStatusCode());
+    }
+
+    public function testPrincipalSetByTheApplicationBindsTheSession(): void
+    {
+        $middleware = new class implements MiddlewareInterface {
+            public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
+            {
+                return $handler->handle($request->withAttribute(StreamableHttpTransport::PRINCIPAL_ATTRIBUTE, $request->getHeaderLine('X-User')));
+            }
+        };
+
+        $initialize = $this->request('POST', $this->initializeRequest(), null)->withHeader('X-User', 'alice');
+        $session = $this->serve($initialize, [$middleware])->getHeaderLine('Mcp-Session-Id');
+
+        $this->assertSame('anonymous', $this->text($this->serve($this->request('POST', $this->toolCall(), null, $session)->withHeader('X-User', 'alice'), [$middleware])));
+        $this->assertSame(404, $this->serve($this->request('POST', $this->toolCall(), null, $session)->withHeader('X-User', 'carol'), [$middleware])->getStatusCode());
+    }
+
+    public function testAccessTokenSetByTheApplicationBindsTheSession(): void
+    {
+        $session = $this->openWith([AccessToken::class => new AccessToken([], ['sub' => 'alice'])]);
+
+        $this->assertSame('alice', $this->text($this->callWith($session, [AccessToken::class => new AccessToken([], ['sub' => 'alice'])])));
+        $this->assertSame(404, $this->callWith($session, [AccessToken::class => new AccessToken([], ['sub' => 'carol'])])->getStatusCode());
+    }
+
+    public function testPrincipalAttributeWinsOverTheAccessToken(): void
+    {
+        $alice = [StreamableHttpTransport::PRINCIPAL_ATTRIBUTE => 'alice'];
+        $session = $this->openWith($alice + [AccessToken::class => new AccessToken([], ['sub' => 'device-1'])]);
+
+        $this->assertSame('device-2', $this->text($this->callWith($session, $alice + [AccessToken::class => new AccessToken([], ['sub' => 'device-2'])])));
+    }
+
+    public function testTokensWithoutSubjectOfAnotherClientAreRefused(): void
+    {
+        $session = $this->openWith([AccessToken::class => new AccessToken([], ['client_id' => 'a'])]);
+
+        $this->assertSame(200, $this->callWith($session, [AccessToken::class => new AccessToken([], ['client_id' => 'a'])])->getStatusCode());
+        $this->assertSame(404, $this->callWith($session, [AccessToken::class => new AccessToken([], ['client_id' => 'b'])])->getStatusCode());
+    }
+
+    public function testTokenWithAClaimThatIsNotUtf8BindsTheSession(): void
+    {
+        $owner = [AccessToken::class => new AccessToken([], ['sub' => "\xB1"])];
+        $other = [AccessToken::class => new AccessToken([], ['sub' => "\xB2"])];
+        $ping = '{"jsonrpc":"2.0","id":2,"method":"ping"}';
+        $session = $this->openWith($owner);
+
+        $this->assertSame(200, $this->serveWith($this->request('POST', $ping, null, $session), $owner)->getStatusCode());
+        $this->assertSame(404, $this->serveWith($this->request('POST', $ping, null, $session), $other)->getStatusCode());
+        $this->assertSame(404, $this->serveWith($this->request('DELETE', '', null, $session), $other)->getStatusCode());
+        $this->assertSame(200, $this->serveWith($this->request('DELETE', '', null, $session), $owner)->getStatusCode());
+    }
+
+    public function testTokenWithoutClaimsDoesNotShareSessionsWithRequestsWithoutToken(): void
+    {
+        $session = $this->openWith([AccessToken::class => new AccessToken()]);
+
+        $this->assertSame(404, $this->callWith($session, [])->getStatusCode());
+    }
+
+    /**
+     * @return iterable<string, array{mixed}>
+     */
+    public static function provideInvalidPrincipals(): iterable
+    {
+        yield 'an integer' => [42];
+        yield 'an empty string' => [''];
+        yield 'a stringable object' => [new class implements \Stringable {
+            public function __toString(): string
+            {
+                return 'alice';
+            }
+        }];
+    }
+
+    #[DataProvider('provideInvalidPrincipals')]
+    public function testInvalidPrincipalAttributeIsRefused(mixed $principal): void
+    {
+        $attributes = [StreamableHttpTransport::PRINCIPAL_ATTRIBUTE => $principal];
+
+        $response = $this->serveWith($this->request('POST', $this->initializeRequest(), null), $attributes);
+        $this->assertSame(500, $response->getStatusCode());
+        $this->assertFalse($response->hasHeader('Mcp-Session-Id'));
+
+        $session = $this->openWith([]);
+        $this->assertSame(500, $this->serveWith($this->request('DELETE', '', null, $session), $attributes)->getStatusCode());
+        $this->assertSame(200, $this->callWith($session, [])->getStatusCode());
+    }
+
+    /**
+     * @return iterable<string, array{string}>
+     */
+    public static function provideInputsRefusedBeforeTheSessionIsChecked(): iterable
+    {
+        yield 'malformed JSON' => ['{'];
+        yield 'an initialize request' => ['{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"client","version":"1.0.0"}}}'];
+    }
+
+    #[DataProvider('provideInputsRefusedBeforeTheSessionIsChecked')]
+    public function testInputRefusedBeforeTheSessionIsCheckedLeavesItsQueueAlone(string $input): void
+    {
+        $protocol = $this->protocol();
+        $session = $this->sessionWithQueuedMessage($protocol);
+
+        $response = $this->post($protocol, $input, $session, 'carol');
+
+        $this->assertStringNotContainsString(self::QUEUED_URI, (string) $response->getBody());
+        $this->assertSame(400, $response->getStatusCode());
+        $this->assertStringContainsString(self::QUEUED_URI, (string) $this->post($protocol, self::PING, $session, 'alice')->getBody());
+    }
+
+    public function testInternalErrorBeforeTheSessionIsCheckedLeavesItsQueueAlone(): void
+    {
+        $session = $this->sessionWithQueuedMessage($this->protocol());
+        $this->store->gcFails = true;
+
+        $response = $this->post($this->protocol(gcProbability: 1), self::PING, $session, 'carol');
+
+        $this->assertStringNotContainsString(self::QUEUED_URI, (string) $response->getBody());
+        $this->assertSame(500, $response->getStatusCode());
+
+        $this->store->gcFails = false;
+        $this->assertStringContainsString(self::QUEUED_URI, (string) $this->post($this->protocol(), self::PING, $session, 'alice')->getBody());
+    }
+
+    public function testSessionOpenedWithoutPrincipalOnlyServesRequestsWithoutOne(): void
+    {
+        $passthrough = [new PassthroughMiddleware()];
+        $session = $this->serve($this->request('POST', $this->initializeRequest(), null), $passthrough)->getHeaderLine('Mcp-Session-Id');
+
+        $this->assertSame('anonymous', $this->text($this->call($session, null, $passthrough)));
+        $this->assertSame(404, $this->call($session, 'alice.ide.1')->getStatusCode());
+    }
+
+    public static function whoami(RequestContext $context): string
+    {
+        return $context->getAccessToken()?->getSubject() ?? 'anonymous';
+    }
+
+    /**
+     * @param array<string, mixed> $attributes
+     */
+    private function openWith(array $attributes): string
+    {
+        $session = $this->serveWith($this->request('POST', $this->initializeRequest(), null), $attributes)->getHeaderLine('Mcp-Session-Id');
+        $this->assertNotSame('', $session);
+
+        return $session;
+    }
+
+    /**
+     * @param array<string, mixed> $attributes
+     */
+    private function callWith(string $session, array $attributes): ResponseInterface
+    {
+        return $this->serveWith($this->request('POST', $this->toolCall(), null, $session), $attributes);
+    }
+
+    /**
+     * @param array<string, mixed> $attributes
+     */
+    private function serveWith(ServerRequestInterface $request, array $attributes): ResponseInterface
+    {
+        foreach ($attributes as $name => $value) {
+            $request = $request->withAttribute($name, $value);
+        }
+
+        return $this->serve($request, [new PassthroughMiddleware()]);
+    }
+
+    private function protocol(int $gcProbability = 0): Protocol
+    {
+        return new Protocol(
+            requestHandlers: [new PingHandler()],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: new SessionManager($this->store, gcProbability: $gcProbability, gcDivisor: 1),
+        );
+    }
+
+    /**
+     * Opens a session of alice, with a message queued outside of any request.
+     */
+    private function sessionWithQueuedMessage(Protocol $protocol): Uuid
+    {
+        $session = new Session($this->store);
+        $session->set('_mcp.principal', hash('sha256', 'alice'));
+        $session->set('initialized', true);
+        $protocol->sendNotification(new ResourceUpdatedNotification(self::QUEUED_URI), $session);
+        $session->save();
+
+        return $session->getId();
+    }
+
+    private function post(Protocol $protocol, string $input, Uuid $session, string $principal): ResponseInterface
+    {
+        $request = $this->request('POST', $input, null, $session->toRfc4122())
+            ->withAttribute(StreamableHttpTransport::PRINCIPAL_ATTRIBUTE, $principal);
+        $transport = new StreamableHttpTransport($request, $this->factory, $this->factory, middleware: [new PassthroughMiddleware()]);
+        $protocol->connect($transport);
+
+        return $transport->listen();
+    }
+
+    private function server(Server $server, ServerRequestInterface $request): ResponseInterface
+    {
+        return $server->run(new StreamableHttpTransport($request, $this->factory, $this->factory, middleware: [$this->authorization()]));
+    }
+
+    private function initialize(string $token): string
+    {
+        $session = $this->serve($this->request('POST', $this->initializeRequest(), $token))->getHeaderLine('Mcp-Session-Id');
+        $this->assertNotSame('', $session);
+
+        return $session;
+    }
+
+    /**
+     * @param list<MiddlewareInterface>|null $middleware
+     */
+    private function call(string $session, ?string $token, ?array $middleware = null): ResponseInterface
+    {
+        return $this->serve($this->request('POST', $this->toolCall(), $token, $session), $middleware);
+    }
+
+    /**
+     * @param list<MiddlewareInterface>|null $middleware
+     */
+    private function serve(ServerRequestInterface $request, ?array $middleware = null): ResponseInterface
+    {
+        return $this->server->run(new StreamableHttpTransport($request, $this->factory, $this->factory, middleware: $middleware ?? [$this->authorization()]));
+    }
+
+    /**
+     * Tokens read `<subject>.<client>.<serial>`: tokens differing only by their serial are refreshes of each other.
+     */
+    private function authorization(): AuthorizationMiddleware
+    {
+        $validator = new class implements AuthorizationTokenValidatorInterface {
+            public function validate(string $accessToken): AuthorizationResult
+            {
+                [$subject, $client] = explode('.', $accessToken);
+
+                return AuthorizationResult::allow(new AccessToken([], ['iss' => 'http://localhost:8180', 'sub' => $subject, 'azp' => $client, 'jti' => $accessToken]));
+            }
+        };
+
+        return new AuthorizationMiddleware(
+            $validator,
+            new ProtectedResourceMetadata('http://localhost/mcp', ['http://localhost:8180']),
+            responseFactory: $this->factory,
+            streamFactory: $this->factory,
+        );
+    }
+
+    private function request(string $method, string $body, ?string $token, ?string $session = null): ServerRequestInterface
+    {
+        $request = $this->factory->createServerRequest($method, 'http://localhost/mcp')
+            ->withHeader('Host', 'localhost')
+            ->withHeader('Content-Type', 'application/json')
+            ->withHeader('Accept', 'application/json, text/event-stream')
+            ->withBody($this->factory->createStream($body));
+
+        if (null !== $token) {
+            $request = $request->withHeader('Authorization', 'Bearer '.$token);
+        }
+
+        if (null !== $session) {
+            $request = $request
+                ->withHeader('Mcp-Session-Id', $session)
+                ->withHeader('MCP-Protocol-Version', ProtocolVersion::V2025_11_25->value);
+        }
+
+        return $request;
+    }
+
+    private function initializeRequest(): string
+    {
+        return json_encode([
+            'jsonrpc' => '2.0',
+            'id' => 1,
+            'method' => 'initialize',
+            'params' => [
+                'protocolVersion' => ProtocolVersion::V2025_11_25->value,
+                'capabilities' => new \stdClass(),
+                'clientInfo' => ['name' => 'client', 'version' => '1.0.0'],
+            ],
+        ], \JSON_THROW_ON_ERROR);
+    }
+
+    private function toolCall(): string
+    {
+        return json_encode([
+            'jsonrpc' => '2.0',
+            'id' => 2,
+            'method' => 'tools/call',
+            'params' => ['name' => 'whoami', 'arguments' => new \stdClass()],
+        ], \JSON_THROW_ON_ERROR);
+    }
+
+    private function text(ResponseInterface $response): string
+    {
+        $body = json_decode((string) $response->getBody(), true, flags: \JSON_THROW_ON_ERROR);
+
+        return $body['result']['content'][0]['text'] ?? 'no result: '.$response->getBody();
+    }
+}
+
+/**
+ * Finds sessions but fails to read them while $unreadable is set, like a cache that is reachable but fails on get().
+ * Fails to look them up while $unavailable is set, like a cache that is down.
+ */
+final class UnreadableSessionStore extends InMemorySessionStore
+{
+    public bool $unreadable = false;
+    public bool $unavailable = false;
+    public bool $gcFails = false;
+
+    public function gc(): array
+    {
+        if ($this->gcFails) {
+            throw new SessionStoreException('Session store is unavailable.');
+        }
+
+        return parent::gc();
+    }
+
+    public function exists(Uuid $id): bool
+    {
+        if ($this->unavailable) {
+            throw new SessionStoreException('Session store is unavailable.');
+        }
+
+        return parent::exists($id);
+    }
+
+    public function read(Uuid $id): string|false
+    {
+        return $this->unreadable ? false : parent::read($id);
+    }
+}
diff --git a/tests/Unit/Server/Transport/TransportShutdownTest.php b/tests/Unit/Server/Transport/TransportShutdownTest.php
new file mode 100644
index 000000000..e969870b6
--- /dev/null
+++ b/tests/Unit/Server/Transport/TransportShutdownTest.php
@@ -0,0 +1,104 @@
+<?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\Transport;
+
+use Mcp\Exception\SessionStoreException;
+use Mcp\Server\Transport\InMemoryTransport;
+use Mcp\Server\Transport\StdioTransport;
+use Mcp\Server\Transport\TransportInterface;
+use Mcp\Tests\Unit\Server\Session\Fixture\RecordingLogger;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Uid\Uuid;
+
+final class TransportShutdownTest extends TestCase
+{
+    private RecordingLogger $logger;
+
+    protected function setUp(): void
+    {
+        $this->logger = new RecordingLogger();
+    }
+
+    public function testInMemoryTransportSurvivesAStoreOutageWhenListening(): void
+    {
+        $transport = new InMemoryTransport(logger: $this->logger);
+        $this->failToEndSession($transport);
+
+        $transport->listen();
+
+        $this->assertStoreOutageIsLogged();
+    }
+
+    public function testInMemoryTransportSurvivesAStoreOutageWhenClosing(): void
+    {
+        $transport = new InMemoryTransport(logger: $this->logger);
+        $this->failToEndSession($transport);
+
+        $transport->close();
+
+        $this->assertStoreOutageIsLogged();
+    }
+
+    public function testStdioTransportSurvivesAStoreOutageWhenListening(): void
+    {
+        $pipes = stream_socket_pair(\STREAM_PF_UNIX, \STREAM_SOCK_STREAM, \STREAM_IPPROTO_IP);
+        $this->assertIsArray($pipes);
+        fclose($pipes[0]);
+        $transport = new StdioTransport($pipes[1], $this->stream(), $this->logger);
+        $this->failToEndSession($transport);
+
+        $this->assertSame(0, $transport->listen());
+
+        $this->assertStoreOutageIsLogged();
+    }
+
+    public function testStdioTransportClosesItsStreamsDespiteAStoreOutage(): void
+    {
+        $input = $this->stream();
+        $output = $this->stream();
+        $transport = new StdioTransport($input, $output, $this->logger);
+        $this->failToEndSession($transport);
+
+        $transport->close();
+
+        $this->assertStoreOutageIsLogged();
+        $this->assertFalse(\is_resource($input));
+        $this->assertFalse(\is_resource($output));
+    }
+
+    /**
+     * @param TransportInterface<mixed> $transport
+     */
+    private function failToEndSession(TransportInterface $transport): void
+    {
+        $transport->setSessionId(Uuid::v4());
+        $transport->onSessionEnd(static function (): bool {
+            throw new SessionStoreException('Session store is unavailable.');
+        });
+    }
+
+    private function assertStoreOutageIsLogged(): void
+    {
+        $this->assertSame(['Session store is unavailable, the session was not ended.'], $this->logger->messagesAt('error'));
+    }
+
+    /**
+     * @return resource
+     */
+    private function stream()
+    {
+        $stream = fopen('php://memory', 'r+');
+        $this->assertIsResource($stream);
+
+        return $stream;
+    }
+}