diff --git a/CHANGELOG.md b/CHANGELOG.md
index f3ee7a5c..3a1cb2f8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,7 +35,8 @@ All notable changes to `mcp/sdk` will be documented in this file.
 * [BC Break] Add `ScopePolicy` as third argument of `AuthorizationMiddleware`, answering `403 insufficient_scope` per method and tool, with scope hierarchies; the `resource_metadata` challenge URL comes from the configured resource instead of the `Host` header.
 * Expose `WWW-Authenticate` in the default `CorsMiddleware`.
 * [BC Break] Fix concurrent Streamable HTTP streams on one session resuming each other's fibers: each stream now polls only the client request its own fiber sent, so an elicitation answer reaches the tool call that asked for it. `Protocol::handleFiberYield()` returns the ID of the request it sent.
-* Fix lost responses on concurrent requests of one session over Streamable HTTP: a POST is answered with its own responses instead of taking them from the session's outgoing queue. Adds `InlineResponseTransportInterface` for transports that answer each request on the exchange that carried it.
+* Fix lost responses on concurrent requests of one session over Streamable HTTP: a POST is answered with its own responses instead of taking them from the session's outgoing queue.
+* [BC Break] `TransportInterface::send()` receives every response of a handler that does not suspend, not only errors raised before a session exists; the session's outgoing queue only carries server-initiated requests and notifications. A custom transport must deliver each message it gets, as a batch gets several.
 * Serve both protocol eras over stdio: `StdioTransport` settles the era on the client's first request and serves `2026-07-28` requests, `subscriptions/listen` and `notifications/cancelled` on the one channel.
 * [BC Break] `StatelessAwareTransportInterface` declares `setHandshakeVersions()`, so a server without the modern era names only the revisions it negotiates when refusing a `2026-07-28` request, e.g. the one set with `Builder::setProtocolVersion()`.
 * Answer a bare `initialize` on a `2026-07-28`-only endpoint with `-32022` naming the served revisions, and a request without a session on the handshake leg with its id.
diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php
index 81cb2957..bbd25019 100644
--- a/src/Server/Protocol.php
+++ b/src/Server/Protocol.php
@@ -16,6 +16,7 @@
 use Mcp\Event\RequestEvent;
 use Mcp\Event\ResponseEvent;
 use Mcp\Exception\InvalidInputMessageException;
+use Mcp\Exception\RuntimeException;
 use Mcp\JsonRpc\MessageFactory;
 use Mcp\Schema\JsonRpc\Error;
 use Mcp\Schema\JsonRpc\Notification;
@@ -32,7 +33,6 @@
 use Mcp\Server\Stateless\RequestStateCodec;
 use Mcp\Server\Suspension\NotificationSuspension;
 use Mcp\Server\Suspension\RequestSuspension;
-use Mcp\Server\Transport\InlineResponseTransportInterface;
 use Mcp\Server\Transport\TransportInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use Psr\Log\LoggerInterface;
@@ -244,7 +244,12 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U
             }
         }
 
-        $session->save();
+        try {
+            $session->save();
+        } catch (\Throwable $e) {
+            // The responses already reached the transport: an error for the same requests would contradict them.
+            $this->logger->error(\sprintf('Failed to save session: %s', $e->getMessage()), ['exception' => $e]);
+        }
     }
 
     /**
@@ -343,6 +348,8 @@ private function handleRequest(TransportInterface $transport, Request $request,
                 $result = $fiber->start();
 
                 if ($fiber->isSuspended()) {
+                    $beforeSuspension = $session->all();
+
                     $awaitedRequestId = null;
                     if ($result instanceof NotificationSuspension) {
                         $this->sendNotification($result->notification, $session);
@@ -350,6 +357,23 @@ private function handleRequest(TransportInterface $transport, Request $request,
                         $awaitedRequestId = $this->sendRequest($result->request, $result->timeout, $session);
                     }
 
+                    // The transport resumes the fiber from what the session holds: it must
+                    // not get the fiber if the request the fiber awaits was not stored. A
+                    // later save must not store that request either, nobody would resume it.
+                    try {
+                        $saved = $session->save();
+                    } catch (\Throwable $e) {
+                        $session->hydrate($beforeSuspension);
+
+                        throw $e;
+                    }
+
+                    if (!$saved) {
+                        $session->hydrate($beforeSuspension);
+
+                        throw new RuntimeException('Failed to save the session of a suspended request.');
+                    }
+
                     $this->trackAwaitedRequest($transport, $awaitedRequestId);
                     $transport->attachFiberToSession($fiber, $session->getId());
 
@@ -481,7 +505,11 @@ public function sendNotification(Notification $notification, SessionInterface $s
     }
 
     /**
-     * Sends a response either immediately or queued for later delivery.
+     * Sends a response through the transport, on the exchange that carried its request.
+     *
+     * Responses never go through the session's outgoing queue: the session is shared by
+     * the concurrent requests of a client and written back whole, so a queued response
+     * could be overwritten by another request, or taken by it.
      *
      * @param TransportInterface<mixed>                            $transport
      * @param Response<ResultInterface|array<string, mixed>>|Error $response
@@ -489,53 +517,41 @@ public function sendNotification(Notification $notification, SessionInterface $s
      */
     private function sendResponse(TransportInterface $transport, Response|Error $response, ?SessionInterface $session, array $context = []): void
     {
-        // Queued in the session, a response can be overwritten or taken by a concurrent
-        // request of the same session: a transport that can answer on the request's
-        // own exchange gets it directly.
-        if (null === $session || $transport instanceof InlineResponseTransportInterface) {
-            $this->logger->debug('Sending immediate response', [
-                'response_id' => $response->getId(),
-            ]);
+        $this->logger->debug('Sending response', [
+            'response_id' => $response->getId(),
+        ]);
 
-            try {
-                $encoded = json_encode($response, \JSON_THROW_ON_ERROR);
-            } catch (\JsonException $e) {
-                $this->logger->error('Failed to encode response to JSON.', [
-                    'message_id' => $response->getId(),
-                    'exception' => $e,
-                ]);
-
-                $fallbackError = new Error(
-                    id: $response->getId(),
-                    code: Error::INTERNAL_ERROR,
-                    message: 'Response could not be encoded to JSON'
-                );
-
-                $encoded = json_encode($fallbackError, \JSON_THROW_ON_ERROR);
-            }
+        try {
+            $encoded = json_encode($response, \JSON_THROW_ON_ERROR);
+        } catch (\JsonException $e) {
+            $this->logger->error('Failed to encode response to JSON.', [
+                'message_id' => $response->getId(),
+                'exception' => $e,
+            ]);
 
-            $context['type'] = 'response';
-            if (null !== $session) {
-                $context['session_id'] = $session->getId();
-            }
+            $fallbackError = new Error(
+                id: $response->getId(),
+                code: Error::INTERNAL_ERROR,
+                message: 'Response could not be encoded to JSON'
+            );
 
-            $transport->send($encoded, $context);
-        } else {
-            $this->logger->info('Queueing server response', [
-                'response_id' => $response->getId(),
-            ]);
+            $encoded = json_encode($fallbackError, \JSON_THROW_ON_ERROR);
+        }
 
-            $this->queueOutgoing($response, ['type' => 'response'], $session);
+        $context['type'] = 'response';
+        if (null !== $session) {
+            $context['session_id'] = $session->getId();
         }
+
+        $transport->send($encoded, $context);
     }
 
     /**
      * Helper to queue outgoing messages in session.
      *
-     * @param Request|Notification|Response<ResultInterface|array<string, mixed>>|Error $message
-     * @param array<string, mixed>                                                      $context
+     * @param array<string, mixed> $context
      */
-    private function queueOutgoing(Request|Notification|Response|Error $message, array $context, SessionInterface $session): void
+    private function queueOutgoing(Request|Notification $message, array $context, SessionInterface $session): void
     {
         try {
             $encoded = json_encode($message, \JSON_THROW_ON_ERROR);
diff --git a/src/Server/Transport/InlineResponseTransportInterface.php b/src/Server/Transport/InlineResponseTransportInterface.php
deleted file mode 100644
index ec4c2277..00000000
--- a/src/Server/Transport/InlineResponseTransportInterface.php
+++ /dev/null
@@ -1,27 +0,0 @@
-<?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\Server\Transport;
-
-/**
- * A transport that answers each request on the exchange that carried it.
- *
- * {@see \Mcp\Server\Protocol} hands such a transport its responses through
- * {@see TransportInterface::send()}, with the session in the `session_id`
- * context key, instead of queueing them in the session. The session is shared
- * by the concurrent requests of a client and written back whole, so a queued
- * response can be overwritten by another request, or taken by it.
- *
- * Server-initiated requests and notifications still go through the session queue.
- */
-interface InlineResponseTransportInterface
-{
-}
diff --git a/src/Server/Transport/StreamableHttpTransport.php b/src/Server/Transport/StreamableHttpTransport.php
index d201b959..847524e1 100644
--- a/src/Server/Transport/StreamableHttpTransport.php
+++ b/src/Server/Transport/StreamableHttpTransport.php
@@ -54,7 +54,7 @@
  *
  * @author Kyrian Obikwelu <koshnawaza@gmail.com>
  */
-class StreamableHttpTransport extends BaseTransport implements StatelessAwareTransportInterface, InlineResponseTransportInterface
+class StreamableHttpTransport extends BaseTransport implements StatelessAwareTransportInterface
 {
     use ReadsBoundedBody;
 
@@ -80,8 +80,8 @@ class StreamableHttpTransport extends BaseTransport implements StatelessAwareTra
     private ?string $immediateResponse = null;
     private ?int $immediateStatusCode = null;
 
-    /** @var list<string> responses to the requests of the current POST, see {@see InlineResponseTransportInterface} */
-    private array $inlineResponses = [];
+    /** @var list<string> responses to the requests of the current POST */
+    private array $responses = [];
 
     /** @var list<MiddlewareInterface>|null null until {@see self::listen()} resolves the defaults */
     private ?array $middleware;
@@ -181,14 +181,15 @@ public function setHandshakeVersions(array $versions): void
 
     public function send(string $data, array $context): void
     {
-        if (isset($context['session_id'])) {
-            $this->inlineResponses[] = $data;
+        // An error with its own HTTP status rejects the whole POST and is its only answer.
+        if (isset($context['status_code'])) {
+            $this->immediateResponse = $data;
+            $this->immediateStatusCode = $context['status_code'];
 
             return;
         }
 
-        $this->immediateResponse = $data;
-        $this->immediateStatusCode = $context['status_code'] ?? 200;
+        $this->responses[] = $data;
     }
 
     public function listen(): ResponseInterface
@@ -222,7 +223,7 @@ protected function handlePostRequest(string $body, ?AccessToken $accessToken = n
         $this->immediateStatusCode = null;
 
         if (null !== $immediateResponse) {
-            $this->inlineResponses = [];
+            $this->responses = [];
 
             return $this->responseFactory->createResponse($immediateStatusCode ?? 200)
                 ->withHeader('Content-Type', 'application/json')
@@ -251,8 +252,8 @@ protected function handleDeleteRequest(): ResponseInterface
 
     protected function createJsonResponse(): ResponseInterface
     {
-        $messages = [...array_column($this->getOutgoingMessages($this->sessionId), 'message'), ...$this->inlineResponses];
-        $this->inlineResponses = [];
+        $messages = [...array_column($this->getOutgoingMessages($this->sessionId), 'message'), ...$this->responses];
+        $this->responses = [];
 
         if ([] === $messages) {
             return $this->responseFactory->createResponse(202)
@@ -277,10 +278,10 @@ protected function createStreamedResponse(): ResponseInterface
         $fiber = $this->sessionFiber;
 
         // The other requests of a batch whose handler did not suspend.
-        $inlineResponses = $this->inlineResponses;
-        $this->inlineResponses = [];
+        $responses = $this->responses;
+        $this->responses = [];
 
-        $callback = function () use ($fiber, $inlineResponses): void {
+        $callback = function () use ($fiber, $responses): void {
             if (null === $fiber) {
                 return;
             }
@@ -288,7 +289,7 @@ protected function createStreamedResponse(): ResponseInterface
             try {
                 $this->logger->info('SSE: Starting request processing loop');
 
-                foreach ($inlineResponses as $message) {
+                foreach ($responses as $message) {
                     echo "event: message\n";
                     echo "data: {$message}\n\n";
                     @ob_flush();
diff --git a/src/Server/Transport/TransportInterface.php b/src/Server/Transport/TransportInterface.php
index 00d8b41f..3066e293 100644
--- a/src/Server/Transport/TransportInterface.php
+++ b/src/Server/Transport/TransportInterface.php
@@ -48,9 +48,12 @@ public function listen(): mixed;
     /**
      * Send a message to the client immediately (bypassing session queue).
      *
-     * Used for session resolution errors when no session is available, and for
-     * every response on a {@see InlineResponseTransportInterface}.
-     * The transport decides HOW to send based on context.
+     * Used for every response of a handler that returns without suspending, on
+     * the exchange that carried its request: a transport must deliver each one,
+     * a batch gets several. A handler that suspends returns its result from the
+     * fiber the transport resumes instead. An error with a `status_code` was
+     * raised before a session was resolved and rejects the whole exchange. The
+     * transport decides HOW to send based on context.
      *
      * @param array<string, mixed> $context Context about this message:
      *                                      - 'session_id': Uuid|null
diff --git a/tests/Unit/Fixtures/RecordingTransport.php b/tests/Unit/Fixtures/RecordingTransport.php
new file mode 100644
index 00000000..0b0bb950
--- /dev/null
+++ b/tests/Unit/Fixtures/RecordingTransport.php
@@ -0,0 +1,28 @@
+<?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\Fixtures;
+
+use Mcp\Server\Transport\InMemoryTransport;
+
+/**
+ * Records what the protocol sends, in the shape of the session's outgoing queue.
+ */
+final class RecordingTransport extends InMemoryTransport
+{
+    /** @var list<array{message: string, context: array<string, mixed>}> */
+    public array $sent = [];
+
+    public function send(string $data, array $context): void
+    {
+        $this->sent[] = ['message' => $data, 'context' => $context];
+    }
+}
diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php
index 89788605..26af24fd 100644
--- a/tests/Unit/Server/ProtocolTest.php
+++ b/tests/Unit/Server/ProtocolTest.php
@@ -34,10 +34,9 @@
 use Mcp\Server\Session\SessionManagerInterface;
 use Mcp\Server\Suspension\NotificationSuspension;
 use Mcp\Server\Suspension\RequestSuspension;
-use Mcp\Server\Transport\InlineResponseTransportInterface;
-use Mcp\Server\Transport\InMemoryTransport;
 use Mcp\Server\Transport\TransportInterface;
 use Mcp\Tests\Unit\Fixtures\PollingLoopTransport;
+use Mcp\Tests\Unit\Fixtures\RecordingTransport;
 use Mcp\Tests\Unit\Fixtures\ThrowingRequest;
 use PHPUnit\Framework\Attributes\DataProvider;
 use PHPUnit\Framework\Attributes\TestDox;
@@ -140,27 +139,11 @@ public function testRequestHandledByFirstMatchingHandler(): void
         $this->sessionManager->method('exists')->willReturn(true);
         $session->method('getId')->willReturn(Uuid::v4());
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
-        // The protocol now queues responses instead of sending them directly
-        // save() is called once during processInput and once during consumeOutgoingMessages
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [$handlerA, $handlerB, $handlerC],
             notificationHandlers: [],
@@ -170,13 +153,13 @@ public function testRequestHandledByFirstMatchingHandler(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}',
             $sessionId
         );
 
-        // Check that the response was queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that the response was sent
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -380,27 +363,11 @@ public function testInvalidMessageStructureReturnsError(): void
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
-        // The protocol now queues responses instead of sending them directly
-        // save() is called once during processInput and once during consumeOutgoingMessages
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [],
             notificationHandlers: [],
@@ -410,13 +377,13 @@ public function testInvalidMessageStructureReturnsError(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "params": {}}',
             $sessionId
         );
 
-        // Check that the error was queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that the error was sent
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -532,6 +499,161 @@ public function testUnexpectedThrowableWhileSavingSessionDoesNotEscape(): void
         );
     }
 
+    #[TestDox('A session that fails to save does not add an error to the responses already sent')]
+    public function testSaveFailureKeepsSentResponses(): void
+    {
+        $handler = $this->createMock(RequestHandlerInterface::class);
+        $handler->method('supports')->willReturn(true);
+        $handler->method('handle')->willReturn(new Response(1, ['status' => 'ok']));
+
+        $session = $this->createMock(SessionInterface::class);
+        $session->method('getId')->willReturn(Uuid::v4());
+        $session->method('save')->willThrowException(new \RuntimeException('storage is gone'));
+
+        $this->sessionManager->method('createWithId')->willReturn($session);
+        $this->sessionManager->method('exists')->willReturn(true);
+
+        $transport = new RecordingTransport();
+
+        $protocol = new Protocol(
+            requestHandlers: [$handler],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $this->sessionManager,
+        );
+
+        $protocol->processInput(
+            $transport,
+            '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "write_file", "arguments": {}}}',
+            Uuid::v4()
+        );
+
+        // The tool ran: answering it with an internal error too would tell the client it did not.
+        $this->assertCount(1, $transport->sent);
+        $this->assertSame(['status' => 'ok'], json_decode($transport->sent[0]['message'], true)['result']);
+    }
+
+    /**
+     * @return iterable<string, array{bool}>
+     */
+    public static function provideSaveFailures(): iterable
+    {
+        yield 'store throws' => [true];
+        yield 'store reports false' => [false];
+    }
+
+    #[TestDox('A suspended handler is not handed to the transport when the session fails to save what it awaits: $_dataName')]
+    #[DataProvider('provideSaveFailures')]
+    public function testSaveFailureAbortsSuspendedRequest(bool $throws): void
+    {
+        $handler = $this->createMock(RequestHandlerInterface::class);
+        $handler->method('supports')->willReturn(true);
+        $handler->method('handle')->willReturnCallback(static function (Request $request, SessionInterface $session): Response {
+            \Fiber::suspend(new RequestSuspension(new PingRequest(), $session->getId()->toRfc4122(), 5));
+
+            return new Response(1, []);
+        });
+
+        $store = new class extends InMemorySessionStore {
+            public bool $failWrites = false;
+            public bool $throws = true;
+
+            public function write(Uuid $id, string $data): bool
+            {
+                if ($this->failWrites && $this->throws) {
+                    throw new \RuntimeException('storage is gone');
+                }
+
+                return !$this->failWrites && parent::write($id, $data);
+            }
+        };
+        $store->throws = $throws;
+        $sessionManager = new SessionManager($store, gcProbability: 0);
+        $sessionId = Uuid::v4();
+        $sessionManager->createWithId($sessionId)->save();
+        $store->failWrites = true;
+
+        // Resumed, the fiber would wait for a client response to a request the session never stored.
+        $this->transport->expects($this->never())->method('attachFiberToSession');
+
+        $sent = [];
+        $this->transport->method('send')->willReturnCallback(static function (string $data) use (&$sent): void {
+            $sent[] = json_decode($data, true);
+        });
+
+        $protocol = new Protocol(
+            requestHandlers: [$handler],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $sessionManager,
+        );
+
+        $protocol->processInput(
+            $this->transport,
+            '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "confirm", "arguments": {}}}',
+            $sessionId
+        );
+
+        $this->assertCount(1, $sent);
+        $this->assertSame(1, $sent[0]['id']);
+        $this->assertSame(Error::INTERNAL_ERROR, $sent[0]['error']['code']);
+    }
+
+    #[TestDox('An aborted suspended request leaves nothing behind when a later save succeeds: $_dataName')]
+    #[DataProvider('provideSaveFailures')]
+    public function testAbortedSuspendedRequestIsRolledBack(bool $throws): void
+    {
+        $handler = $this->createMock(RequestHandlerInterface::class);
+        $handler->method('supports')->willReturn(true);
+        $handler->method('handle')->willReturnCallback(static function (Request $request, SessionInterface $session): Response {
+            \Fiber::suspend(new RequestSuspension(new PingRequest(), $session->getId()->toRfc4122(), 5));
+
+            return new Response(1, []);
+        });
+
+        // Fails once, then recovers: the save at the end of processing goes through.
+        $store = new class extends InMemorySessionStore {
+            public int $failingWrites = 0;
+            public bool $throws = true;
+
+            public function write(Uuid $id, string $data): bool
+            {
+                if ($this->failingWrites > 0) {
+                    --$this->failingWrites;
+                    if ($this->throws) {
+                        throw new \RuntimeException('storage hiccup');
+                    }
+
+                    return false;
+                }
+
+                return parent::write($id, $data);
+            }
+        };
+        $store->throws = $throws;
+        $sessionManager = new SessionManager($store, gcProbability: 0);
+        $sessionId = Uuid::v4();
+        $sessionManager->createWithId($sessionId)->save();
+        $store->failingWrites = 1;
+
+        $protocol = new Protocol(
+            requestHandlers: [$handler],
+            notificationHandlers: [],
+            messageFactory: MessageFactory::make(),
+            sessionManager: $sessionManager,
+        );
+
+        $protocol->processInput(
+            $this->transport,
+            '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "confirm", "arguments": {}}}',
+            $sessionId
+        );
+
+        // The client must not get a request for a tool call that was already answered with an error.
+        $this->assertSame([], $protocol->consumeOutgoingMessages($sessionId));
+        $this->assertSame([], $protocol->getPendingRequests($sessionId));
+    }
+
     #[TestDox('A failing notification event listener does not produce a response')]
     public function testFailingNotificationListenerDoesNotProduceResponse(): void
     {
@@ -598,21 +720,7 @@ public function testInvalidMessagePreservesRecoverableId(string $input, string|i
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // Configure session mock for queue operations (mirrors testInvalidMessageStructureReturnsError).
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
+        $transport = new RecordingTransport();
 
         $protocol = new Protocol(
             requestHandlers: [],
@@ -625,12 +733,12 @@ public function testInvalidMessagePreservesRecoverableId(string $input, string|i
         // Valid JSON carrying a real id but missing method/result/error: the message is
         // structurally invalid, yet its id IS recoverable from the decoded payload.
         $protocol->processInput(
-            $this->transport,
+            $transport,
             $input,
             $sessionId
         );
 
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -647,27 +755,11 @@ public function testRequestWithoutHandlerReturnsMethodNotFoundError(): void
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
-        // The protocol now queues responses instead of sending them directly
-        // save() is called once during processInput and once during consumeOutgoingMessages
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [],
             notificationHandlers: [],
@@ -677,13 +769,13 @@ public function testRequestWithoutHandlerReturnsMethodNotFoundError(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "id": 1, "method": "ping"}',
             $sessionId
         );
 
-        // Check that the error was queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that the error was sent
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -704,27 +796,11 @@ public function testHandlerInvalidArgumentReturnsInvalidParamsError(): void
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
-        // The protocol now queues responses instead of sending them directly
-        // save() is called once during processInput and once during consumeOutgoingMessages
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [$handler],
             notificationHandlers: [],
@@ -734,13 +810,13 @@ public function testHandlerInvalidArgumentReturnsInvalidParamsError(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "test"}}',
             $sessionId
         );
 
-        // Check that the error was queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that the error was sent
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -761,27 +837,11 @@ public function testHandlerUnexpectedExceptionReturnsInternalError(): void
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
-        // The protocol now queues responses instead of sending them directly
-        // save() is called once during processInput and once during consumeOutgoingMessages
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [$handler],
             notificationHandlers: [],
@@ -791,13 +851,13 @@ public function testHandlerUnexpectedExceptionReturnsInternalError(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "test"}}',
             $sessionId
         );
 
-        // Check that the error was queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that the error was sent
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -822,6 +882,11 @@ public function testOutboundRequestFailureIsAnsweredUnderInboundRequestId(): voi
         $exception = new \RuntimeException('Transport unavailable');
         $this->transport->method('attachFiberToSession')->willThrowException($exception);
 
+        $sent = [];
+        $this->transport->method('send')->willReturnCallback(static function (string $data) use (&$sent): void {
+            $sent[] = json_decode($data, true);
+        });
+
         $sessionManager = new SessionManager(new InMemorySessionStore());
 
         $events = [];
@@ -854,14 +919,14 @@ public function testOutboundRequestFailureIsAnsweredUnderInboundRequestId(): voi
         $this->assertSame($exception, $errorEvents[0]->getThrowable());
         $this->assertSame(1, $errorEvents[0]->getError()->getId());
 
+        // The outbound request was queued for the client before the failure; the error is the response.
         $outgoing = array_map(static fn (array $outgoingMessage): array => json_decode($outgoingMessage['message'], true), $protocol->consumeOutgoingMessages($sessionId));
-        $outbound = array_values(array_filter($outgoing, static fn (array $message): bool => 'ping' === ($message['method'] ?? null)));
-        $this->assertCount(1, $outbound);
+        $this->assertCount(1, $outgoing);
+        $this->assertSame('ping', $outgoing[0]['method']);
 
-        $errors = array_values(array_filter($outgoing, static fn (array $message): bool => isset($message['error'])));
-        $this->assertCount(1, $errors);
-        $this->assertSame(1, $errors[0]['id']);
-        $this->assertSame(Error::INTERNAL_ERROR, $errors[0]['error']['code']);
+        $this->assertCount(1, $sent);
+        $this->assertSame(1, $sent[0]['id']);
+        $this->assertSame(Error::INTERNAL_ERROR, $sent[0]['error']['code']);
     }
 
     #[TestDox('Concurrent streams on one session each poll only the client request their own fiber sent')]
@@ -995,27 +1060,11 @@ public function testSuccessfulRequestReturnsResponseWithSessionId(): void
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
-        // The protocol now queues responses instead of sending them directly
-        // save() is called once during processInput and once during consumeOutgoingMessages
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [$handler],
             notificationHandlers: [],
@@ -1024,13 +1073,13 @@ public function testSuccessfulRequestReturnsResponseWithSessionId(): void
         );
 
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}',
             $sessionId
         );
 
-        // Check that the response was queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that the response was sent
+        $outgoing = $transport->sent;
         $this->assertCount(1, $outgoing);
 
         $message = json_decode($outgoing[0]['message'], true);
@@ -1038,8 +1087,8 @@ public function testSuccessfulRequestReturnsResponseWithSessionId(): void
         $this->assertEquals(['status' => 'ok'], $message['result']);
     }
 
-    #[TestDox('An inline response transport gets the response directly, not through the session queue')]
-    public function testInlineResponseTransportGetsResponseDirectly(): void
+    #[TestDox('A response goes to the transport with its session, never through the session queue')]
+    public function testResponseIsSentNotQueued(): void
     {
         $handler = $this->createMock(RequestHandlerInterface::class);
         $handler->method('supports')->willReturn(true);
@@ -1049,15 +1098,7 @@ public function testInlineResponseTransportGetsResponseDirectly(): void
         $sessionId = Uuid::v4();
         $sessions->createWithId($sessionId)->save();
 
-        $transport = new class extends InMemoryTransport implements InlineResponseTransportInterface {
-            /** @var list<array{string, array<string, mixed>}> */
-            public array $sent = [];
-
-            public function send(string $data, array $context): void
-            {
-                $this->sent[] = [$data, $context];
-            }
-        };
+        $transport = new RecordingTransport();
 
         $protocol = new Protocol(
             requestHandlers: [$handler],
@@ -1073,10 +1114,9 @@ public function send(string $data, array $context): void
         );
 
         $this->assertCount(1, $transport->sent);
-        [$data, $context] = $transport->sent[0];
-        $this->assertSame(['status' => 'ok'], json_decode($data, true)['result']);
-        $this->assertSame('response', $context['type']);
-        $this->assertEquals($sessionId, $context['session_id']);
+        $this->assertSame(['status' => 'ok'], json_decode($transport->sent[0]['message'], true)['result']);
+        $this->assertSame('response', $transport->sent[0]['context']['type']);
+        $this->assertEquals($sessionId, $transport->sent[0]['context']['session_id']);
         $this->assertSame([], $protocol->consumeOutgoingMessages($sessionId));
     }
 
@@ -1096,29 +1136,14 @@ public function testBatchRequestsAreProcessed(): void
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
 
-        // Configure session mock for queue operations
-        $queue = [];
-        $session->method('get')->willReturnCallback(static function ($key, $default = null) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                return $queue;
-            }
-
-            return $default;
-        });
-
-        $session->method('set')->willReturnCallback(static function ($key, $value) use (&$queue) {
-            if ('_mcp.outgoing_queue' === $key) {
-                $queue = $value;
-            }
-        });
-
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
-        // The protocol now queues responses instead of sending them directly
-        $session->expects($this->exactly(2))
+        $session->expects($this->once())
             ->method('save');
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [$handlerA],
             notificationHandlers: [],
@@ -1128,13 +1153,13 @@ public function testBatchRequestsAreProcessed(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '[{"jsonrpc": "2.0", "method": "tools/list", "id": 1}, {"jsonrpc": "2.0", "method": "prompts/list", "id": 2}]',
             $sessionId
         );
 
-        // Check that both responses were queued in the session
-        $outgoing = $protocol->consumeOutgoingMessages($sessionId);
+        // Check that both responses were sent
+        $outgoing = $transport->sent;
         $this->assertCount(2, $outgoing);
 
         foreach ($outgoing as $outgoingMessage) {
@@ -1391,8 +1416,6 @@ public function testResponseEventIsDispatched(): void
     #[TestDox('ResponseEvent modification is used when sending')]
     public function testResponseEventModificationIsUsed(): void
     {
-        $outgoingQueue = [];
-
         $eventDispatcher = $this->createMock(EventDispatcherInterface::class);
         $eventDispatcher
             ->method('dispatch')
@@ -1416,17 +1439,12 @@ public function testResponseEventModificationIsUsed(): void
         $session = $this->createMock(SessionInterface::class);
         $session->method('getId')->willReturn(Uuid::v4());
         $session->method('get')->willReturn([]);
-        $session
-            ->method('set')
-            ->willReturnCallback(static function ($key, $value) use (&$outgoingQueue) {
-                if ('_mcp.outgoing_queue' === $key) {
-                    $outgoingQueue = $value;
-                }
-            });
 
         $this->sessionManager->method('createWithId')->willReturn($session);
         $this->sessionManager->method('exists')->willReturn(true);
 
+        $transport = new RecordingTransport();
+
         $protocol = new Protocol(
             requestHandlers: [$handler],
             notificationHandlers: [],
@@ -1437,18 +1455,15 @@ public function testResponseEventModificationIsUsed(): void
 
         $sessionId = Uuid::v4();
         $protocol->processInput(
-            $this->transport,
+            $transport,
             '{"jsonrpc": "2.0", "method": "ping", "id": 1}',
             $sessionId
         );
 
-        // Verify the MODIFIED response was queued
-        $this->assertNotEmpty($outgoingQueue);
-        $lastQueued = end($outgoingQueue);
-        $this->assertIsArray($lastQueued);
-        $this->assertArrayHasKey('message', $lastQueued);
+        // Verify the MODIFIED response was sent
+        $this->assertCount(1, $transport->sent);
 
-        $decoded = json_decode($lastQueued['message'], true);
+        $decoded = json_decode($transport->sent[0]['message'], true);
         $this->assertSame('modified', $decoded['result']['result']);
         $this->assertFalse($decoded['result']['original']);
     }
diff --git a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php
index 2e0a7a28..5950960e 100644
--- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php
+++ b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php
@@ -501,7 +501,7 @@ public function testConcurrentPostsOfOneSessionEachGetTheirOwnResponse(bool $rea
     }
 
     #[TestDox('a batch streamed over SSE still carries the responses that did not suspend')]
-    public function testStreamedBatchCarriesInlineResponses(): void
+    public function testStreamedBatchCarriesItsOtherResponses(): void
     {
         $store = new InterleavingSessionStore();
         $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}')
@@ -528,7 +528,7 @@ public function testStreamedBatchCarriesInlineResponses(): void
     }
 
     #[TestDox('a batch answered as JSON carries the queued notifications first, then its responses, in one array')]
-    public function testJsonBatchCarriesQueuedNotificationsAndInlineResponses(): void
+    public function testJsonBatchCarriesQueuedNotificationsAndItsResponses(): void
     {
         $store = new InterleavingSessionStore();
         $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}')