diff --git a/src/Client.php b/src/Client.php
index 149b5d1..f4e7cfa 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -4,7 +4,6 @@
 
 use MongoDB\BSON\Document;
 use MongoDB\BSON\Int64;
-use MongoDB\Driver\Exception\InvalidArgumentException;
 use Ramsey\Uuid\Uuid;
 use stdClass;
 use Swoole\Client as SwooleClient;
@@ -253,7 +252,8 @@ public function connect(): self
         if ($this->port <= 0 || $this->port > 65535) {
             throw new Exception('MongoDB port must be between 1 and 65535');
         }
-        if (!$this->client->connect($this->host, $this->port)) {
+        if (!$this->client->connect($this->host, $this->port, $this->timeout)) {
+            $this->invalidate();
             throw new Exception("Failed to connect to MongoDB at {$this->host}:{$this->port}");
         }
 
@@ -430,17 +430,33 @@ public function send(mixed $data): stdClass|array|int
             $this->connect();
         }
 
+        $length = \strlen($data);
         $result = $this->client->send($data);
 
-        // If send fails, try to reconnect once
-        if ($result === false) {
+        // A partial write corrupts the request boundary just like a failed write.
+        if ($result !== $length) {
             $errCode = $this->client->errCode ?? 0;
-            $this->close();
-            $this->connect();
+            $sessions = $this->sessions;
+            $this->invalidate();
+
+            try {
+                $this->connect();
+            } catch (\Throwable $error) {
+                if ($this->isConnected || $this->client->isConnected()) {
+                    $this->invalidate();
+                }
+
+                throw $error;
+            }
+
+            // Logical sessions survive only after a fresh socket authenticates.
+            $this->sessions = $sessions;
             $result = $this->client->send($data);
-            if ($result === false) {
+
+            if ($result !== $length) {
                 // Prefer the first non-zero errCode (retry may report 0).
                 $errCode = ($this->client->errCode ?? 0) ?: $errCode;
+                $this->invalidate();
                 throw new Exception(
                     'Failed to send data to MongoDB after reconnection attempt'
                     . ($errCode !== 0 ? " (errCode={$errCode})" : ''),
@@ -476,6 +492,7 @@ private function receive(): stdClass|array|int
 
         do {
             if (\microtime(true) >= $deadline) {
+                $this->invalidate();
                 throw new Exception('Receive timeout: no data received within reasonable time', 11601);
             }
 
@@ -485,6 +502,7 @@ private function receive(): stdClass|array|int
             // false => socket-level wait already elapsed with no payload.
             // Do not retry: that turns one timeout into thousands.
             if ($chunk === false) {
+                $this->invalidate();
                 throw new Exception(
                     'Receive timeout: no data received within reasonable time'
                     . ($errCode !== 0 ? " (errCode={$errCode})" : ''),
@@ -495,6 +513,7 @@ private function receive(): stdClass|array|int
             // Some Swoole builds return "" on recv timeout instead of false.
             // errCode != 0 distinguishes that from a transient empty read.
             if ($chunk === '' && $errCode !== 0) {
+                $this->invalidate();
                 throw new Exception(
                     'Receive timeout: no data received within reasonable time'
                     . " (errCode={$errCode})",
@@ -532,16 +551,23 @@ private function receive(): stdClass|array|int
 
                 // Validate response length before allocating memory to prevent memory exhaustion
                 if ($responseLength > 16777216) { // 16MB limit
+                    $this->invalidate();
                     throw new Exception('Response too large: ' . $responseLength . ' bytes');
                 }
 
                 // Validate for negative or tiny values
                 if ($responseLength < 21) { // Minimum MongoDB message size
+                    $this->invalidate();
                     throw new Exception('Invalid response length: ' . $responseLength . ' bytes');
                 }
             }
 
-            if ($responseLength !== null && $receivedLength >= $responseLength) {
+            if ($responseLength !== null && $receivedLength > $responseLength) {
+                $this->invalidate();
+                throw new Exception('Response length mismatch');
+            }
+
+            if ($responseLength !== null && $receivedLength === $responseLength) {
                 break;
             }
         } while (true);
@@ -1657,10 +1683,37 @@ public function close(): void
             $this->client->close();
         }
 
+        $this->reset();
+    }
+
+    /**
+     * Physically close a failed socket without writing session cleanup to it.
+     */
+    private function invalidate(): void
+    {
+        try {
+            if ($this->client instanceof CoroutineClient) {
+                @$this->client->close();
+            } else {
+                @$this->client->close(true);
+            }
+        } catch (\Throwable) {
+            // The transport may already be closed or only partially initialized.
+        } finally {
+            $this->reset();
+        }
+    }
+
+    /**
+     * Clear all state associated with the current physical connection.
+     */
+    private function reset(): void
+    {
         $this->isConnected = false;
         $this->sessions = [];
         $this->clusterTime = null;
         $this->operationTime = null;
+        $this->replicaSet = null;
     }
 
     /**
@@ -1686,7 +1739,13 @@ private function parseResponse(string $response, int $responseLength): stdClass|
          * These 21 bytes are protocol metadata and precede the actual BSON-encoded document in the response.
          */
 
-        if (\strlen($response) < 21) {
+        if (\strlen($response) !== $responseLength) {
+            $this->invalidate();
+            throw new Exception('Response length mismatch');
+        }
+
+        if ($responseLength < 21) {
+            $this->invalidate();
             throw new Exception('Invalid response: too short');
         }
 
@@ -1695,19 +1754,32 @@ private function parseResponse(string $response, int $responseLength): stdClass|
         $headerData = \unpack('VmessageLength/VrequestID/VresponseTo/VopCode', $header);
 
         // Validate message length
-        if ($headerData['messageLength'] !== $responseLength) {
+        if ($headerData === false || $headerData['messageLength'] !== $responseLength) {
+            $this->invalidate();
             throw new Exception('Response length mismatch');
         }
 
-        // Extract flag bits and payload type
-        $flagBits = \unpack('V', \substr($response, 16, 4))[1];
+        // query() emits OP_MSG and this parser decodes its 21-byte envelope.
+        // OP_REPLY uses an incompatible 36-byte envelope and is not supported.
+        if ($headerData['opCode'] !== 2013) {
+            $this->invalidate();
+            throw new Exception('Invalid response operation code: ' . $headerData['opCode']);
+        }
+
+        // Extract payload type after the flag bits.
         $payloadType = \ord(\substr($response, 20, 1));
 
+        if ($payloadType !== 0) {
+            $this->invalidate();
+            throw new Exception('Invalid response payload type: ' . $payloadType);
+        }
+
         // Extract BSON document (skip header + flagBits + payloadType)
         $bsonString = \substr($response, 21, $responseLength - 21);
 
         if (empty($bsonString)) {
-            return new \stdClass();
+            $this->invalidate();
+            throw new Exception('Invalid response: missing BSON document');
         }
 
         try {
@@ -1718,45 +1790,45 @@ private function parseResponse(string $response, int $responseLength): stdClass|
             if (\is_array($result)) {
                 $result = (object)$result;
             }
+        } catch (\Throwable $error) {
+            $this->invalidate();
+            throw new Exception('Failed to parse BSON response: ' . $error->getMessage(), 0, $error);
+        }
 
-            // Check for write errors (duplicate key, etc.)
-            if (\property_exists($result, 'writeErrors') && !empty($result->writeErrors)) {
-                throw new Exception(
-                    $result->writeErrors[0]->errmsg,
-                    $result->writeErrors[0]->code
-                );
-            }
+        // These are fully decoded MongoDB command errors, not transport corruption.
+        if (\property_exists($result, 'writeErrors') && !empty($result->writeErrors)) {
+            $writeError = $result->writeErrors[0] ?? null;
 
-            // Check for general MongoDB errors
-            if (\property_exists($result, 'errmsg')) {
-                throw new Exception(
-                    'E' . $result->code . ' ' . $result->codeName . ': ' . $result->errmsg,
-                    $result->code
-                );
+            if (\is_object($writeError) && isset($writeError->errmsg, $writeError->code)) {
+                throw new Exception($writeError->errmsg, $writeError->code);
             }
 
-            // Check for operation success
-            if (\property_exists($result, 'n') && $result->ok === 1.0) {
-                return $result->n;
-            }
+            $this->invalidate();
+            throw new Exception('Invalid write error response');
+        }
 
-            if (\property_exists($result, 'nonce') && $result->ok === 1.0) {
-                return $result;
-            }
+        if (\property_exists($result, 'errmsg')) {
+            $code = (int)($result->code ?? 0);
+            $name = (string)($result->codeName ?? 'MongoError');
 
-            if ($result->ok === 1.0) {
-                return $result;
-            }
+            throw new Exception('E' . $code . ' ' . $name . ': ' . $result->errmsg, $code);
+        }
 
-            return $result->cursor->firstBatch;
-        } catch (InvalidArgumentException $e) {
-            throw new Exception('Failed to parse BSON response: ' . $e->getMessage());
-        } catch (\Exception $e) {
-            if ($e instanceof Exception) {
-                throw $e;
+        if (!\property_exists($result, 'ok')) {
+            $this->invalidate();
+            throw new Exception('Invalid response: missing operation status');
+        }
+
+        if ($result->ok === 1.0) {
+            if (\property_exists($result, 'n')) {
+                return $result->n;
             }
-            throw new Exception('Error parsing response: ' . $e->getMessage());
+
+            return $result;
         }
+
+        $this->invalidate();
+        throw new Exception('Invalid unsuccessful response');
     }
 
     /**
diff --git a/tests/ClientTest.php b/tests/ClientTest.php
new file mode 100644
index 0000000..4cab1b3
--- /dev/null
+++ b/tests/ClientTest.php
@@ -0,0 +1,582 @@
+<?php
+
+namespace Utopia\Tests;
+
+use MongoDB\BSON\Binary;
+use MongoDB\BSON\Document;
+use PHPUnit\Framework\TestCase;
+use ReflectionMethod;
+use ReflectionProperty;
+use Swoole\Client as SwooleClient;
+use Swoole\Coroutine\Client as CoroutineClient;
+use Utopia\Mongo\Auth;
+use Utopia\Mongo\Client;
+use Utopia\Mongo\Exception;
+
+trait TransportDouble
+{
+    public bool $open = true;
+    public array $connectionResults = [];
+    public array $connects = [];
+    public array $events = [];
+    public array $receives = [];
+    public array $sends = [];
+    public array $writes = [];
+
+    private function connectTransport(string $host, int $port, float $timeout, int $flags): bool
+    {
+        $this->events[] = ['connect', $host, $port, $timeout, $flags];
+        $this->connects[] = [$host, $port, $timeout, $flags];
+        $result = array_shift($this->connectionResults) ?? true;
+        $this->open = $result;
+
+        return $result;
+    }
+
+    private function receiveTransport(): string|false
+    {
+        $this->events[] = ['receive'];
+        $receive = array_shift($this->receives) ?? ['result' => '', 'error' => 0];
+        $this->errCode = $receive['error'];
+
+        return $receive['result'];
+    }
+
+    private function sendTransport(string $data): int|false
+    {
+        $this->events[] = ['send', $data];
+        $this->writes[] = $data;
+        $send = array_shift($this->sends) ?? ['result' => strlen($data), 'error' => 0];
+        $this->errCode = $send['error'];
+
+        return $send['result'];
+    }
+}
+
+final class SyncTransportDouble extends SwooleClient
+{
+    use TransportDouble;
+
+    public array $closes = [];
+
+    public function __construct()
+    {
+    }
+
+    public function close(bool $force = false): bool
+    {
+        $this->events[] = ['close', $force];
+        $this->closes[] = [$force];
+        $this->open = false;
+
+        return true;
+    }
+
+    public function connect(string $host, int $port = 0, float $timeout = 0.5, int $sock_flag = 0): bool
+    {
+        return $this->connectTransport($host, $port, $timeout, $sock_flag);
+    }
+
+    public function isConnected(): bool
+    {
+        return $this->open;
+    }
+
+    public function recv(int $size = 65535, int $flag = 0): string|false
+    {
+        return $this->receiveTransport();
+    }
+
+    public function send(string $data, int $flag = 0): int|false
+    {
+        return $this->sendTransport($data);
+    }
+}
+
+final class CoroutineTransportDouble extends CoroutineClient
+{
+    use TransportDouble;
+
+    public int $closes = 0;
+
+    public function __construct()
+    {
+    }
+
+    public function close(): bool
+    {
+        $this->events[] = ['close'];
+        $this->closes++;
+        $this->open = false;
+
+        return true;
+    }
+
+    public function connect(string $host, int $port = 0, float $timeout = 0.5, int $sock_flag = 0): bool
+    {
+        return $this->connectTransport($host, $port, $timeout, $sock_flag);
+    }
+
+    public function isConnected(): bool
+    {
+        return $this->open;
+    }
+
+    public function recv(float $timeout = 0): string|false
+    {
+        return $this->receiveTransport();
+    }
+
+    public function send(string $data, float $timeout = 0): int|false
+    {
+        return $this->sendTransport($data);
+    }
+}
+
+final class AuthenticationDouble extends Auth
+{
+    public function __construct()
+    {
+    }
+
+    public function continue($data): array
+    {
+        return [['ping' => 1], 'admin'];
+    }
+
+    public function start(): array
+    {
+        return [['ping' => 1], 'admin'];
+    }
+}
+
+final class ReconnectingClient extends Client
+{
+    public SwooleClient|CoroutineClient $transport;
+
+    public function connect(): self
+    {
+        $this->transport->connect('mongo', 27017, 30.0);
+        $connection = new ReflectionProperty(Client::class, 'isConnected');
+        $connection->setAccessible(true);
+        $connection->setValue($this, true);
+
+        return $this;
+    }
+}
+
+final class ClientTest extends TestCase
+{
+    public function testConnectPassesConfiguredTimeout(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->open = false;
+        $transport->receives = [
+            ['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
+            ['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
+        ];
+        $client = $this->client($transport, timeout: 7.5);
+        $this->set($client, 'auth', new AuthenticationDouble());
+
+        $client->connect();
+
+        $this->assertSame([['mongo', 27017, 7.5, 0]], $transport->connects);
+    }
+
+    public function testSyncReceiveFailureHardClosesAndClearsState(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [['result' => false, 'error' => 11]];
+        $client = $this->client($transport);
+        $this->seedState($client);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertSame(11601, $exception->getCode());
+        $this->assertStringContainsString('errCode=11', $exception->getMessage());
+        $this->assertSame([[true]], $transport->closes);
+        $this->assertSame([], $transport->connects);
+        $this->assertStateCleared($client);
+    }
+
+    public function testCoroutineReceiveFailureHardClosesAndClearsState(): void
+    {
+        $transport = new CoroutineTransportDouble();
+        $transport->receives = [['result' => false, 'error' => 11]];
+        $client = $this->client($transport);
+        $this->seedState($client);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertSame(11601, $exception->getCode());
+        $this->assertStringContainsString('errCode=11', $exception->getMessage());
+        $this->assertSame(1, $transport->closes);
+        $this->assertSame([], $transport->connects);
+        $this->assertStateCleared($client);
+    }
+
+    public function testTransientEmptyReceiveContinuesWithoutClosing(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [
+            ['result' => '', 'error' => 0],
+            ['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
+        ];
+        $client = $this->client($transport);
+
+        $result = $this->receive($client);
+
+        $this->assertSame(1.0, $result->ok);
+        $this->assertSame([], $transport->closes);
+    }
+
+    public function testErroredEmptyReceiveHardClosesTransport(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [['result' => '', 'error' => 11]];
+        $client = $this->client($transport);
+        $this->seedState($client);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertSame(11601, $exception->getCode());
+        $this->assertStringContainsString('errCode=11', $exception->getMessage());
+        $this->assertSame([[true]], $transport->closes);
+        $this->assertStateCleared($client);
+    }
+
+    public function testReceiveDeadlineHardClosesTransport(): void
+    {
+        $transport = new SyncTransportDouble();
+        $client = $this->client($transport, timeout: 0.0001);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertSame(11601, $exception->getCode());
+        $this->assertSame([[true]], $transport->closes);
+    }
+
+    public function testSyncSendFailureHardClosesBeforeFreshFullMessageRetry(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->sends = [['result' => false, 'error' => 11]];
+        $transport->receives = [['result' => $this->frame(['ok' => 1.0]), 'error' => 0]];
+        $client = $this->reconnectingClient($transport);
+        $this->seedConnectionContext($client);
+
+        $result = $client->send('complete-message');
+
+        $this->assertSame(1.0, $result->ok);
+        $this->assertSame(['complete-message', 'complete-message'], $transport->writes);
+        $this->assertSame(
+            ['send', 'close', 'connect', 'send', 'receive'],
+            array_column($transport->events, 0)
+        );
+        $this->assertSame([[true]], $transport->closes);
+        $this->assertSame([], $this->get($client, 'sessions'));
+        $this->assertConnectionContextCleared($client);
+        $this->assertTrue($this->get($client, 'isConnected'));
+    }
+
+    public function testSuccessfulSendRecoveryPreservesTransactionSessionForCommit(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [[
+            'result' => $this->frame([
+                'ok' => 1.0,
+                'id' => ['id' => new Binary(str_repeat("\1", 16), Binary::TYPE_GENERIC)],
+            ]),
+            'error' => 0,
+        ]];
+        $client = $this->reconnectingClient($transport);
+        $this->set($client, 'isConnected', true);
+        $session = $client->startSession();
+        $this->assertTrue($client->startTransaction($session));
+
+        $transport->sends = [['result' => false, 'error' => 11]];
+        $transport->receives = [
+            ['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
+            ['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
+            ['result' => $this->frame(['ok' => 1.0]), 'error' => 0],
+        ];
+
+        $operation = $client->query(['ping' => 1, 'session' => $session]);
+        $commit = $client->commitTransaction($session);
+        $sessions = $this->get($client, 'sessions');
+        $ended = $client->endSessions([$session]);
+
+        $this->assertSame(1.0, $operation->ok);
+        $this->assertSame(1.0, $commit->ok);
+        $this->assertSame(1.0, $ended->ok);
+        $this->assertSame(Client::TRANSACTION_COMMITTED, $sessions[$session['sessionId']]['state']);
+        $this->assertSame([], $this->get($client, 'sessions'));
+    }
+
+    public function testCoroutineSendFailureHardClosesBeforeFreshFullMessageRetry(): void
+    {
+        $transport = new CoroutineTransportDouble();
+        $transport->sends = [['result' => false, 'error' => 11]];
+        $transport->receives = [['result' => $this->frame(['ok' => 1.0]), 'error' => 0]];
+        $client = $this->reconnectingClient($transport);
+
+        $result = $client->send('complete-message');
+
+        $this->assertSame(1.0, $result->ok);
+        $this->assertSame(['complete-message', 'complete-message'], $transport->writes);
+        $this->assertSame(
+            ['send', 'close', 'connect', 'send', 'receive'],
+            array_column($transport->events, 0)
+        );
+        $this->assertSame(1, $transport->closes);
+    }
+
+    public function testSyncShortSendHardClosesBeforeRetry(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->sends = [['result' => 4, 'error' => 0]];
+        $transport->receives = [['result' => $this->frame(['ok' => 1.0]), 'error' => 0]];
+        $client = $this->reconnectingClient($transport);
+
+        $client->send('complete-message');
+
+        $this->assertSame([[true]], $transport->closes);
+        $this->assertSame(['complete-message', 'complete-message'], $transport->writes);
+    }
+
+    public function testCoroutineShortSendHardClosesBeforeRetry(): void
+    {
+        $transport = new CoroutineTransportDouble();
+        $transport->sends = [['result' => 4, 'error' => 0]];
+        $transport->receives = [['result' => $this->frame(['ok' => 1.0]), 'error' => 0]];
+        $client = $this->reconnectingClient($transport);
+
+        $client->send('complete-message');
+
+        $this->assertSame(1, $transport->closes);
+        $this->assertSame(['complete-message', 'complete-message'], $transport->writes);
+    }
+
+    public function testRetryFailureHardClosesReplacementTransport(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->sends = [
+            ['result' => false, 'error' => 11],
+            ['result' => false, 'error' => 0],
+        ];
+        $client = $this->reconnectingClient($transport);
+        $this->seedState($client);
+
+        try {
+            $client->send('complete-message');
+            $this->fail('Expected send failure');
+        } catch (Exception $exception) {
+            $this->assertSame(11, $exception->getCode());
+        }
+
+        $this->assertSame([[true], [true]], $transport->closes);
+        $this->assertStateCleared($client);
+    }
+
+    public function testRetryShortSendHardClosesReplacementTransport(): void
+    {
+        $transport = new CoroutineTransportDouble();
+        $transport->sends = [
+            ['result' => false, 'error' => 11],
+            ['result' => 4, 'error' => 0],
+        ];
+        $transport->receives = [['result' => $this->frame(['ok' => 1.0]), 'error' => 0]];
+        $client = $this->reconnectingClient($transport);
+        $this->seedState($client);
+
+        try {
+            $client->send('complete-message');
+            $this->fail('Expected send failure');
+        } catch (Exception $exception) {
+            $this->assertSame(11, $exception->getCode());
+        }
+
+        $this->assertSame(2, $transport->closes);
+        $this->assertStateCleared($client);
+    }
+
+    public function testFailedReconnectDoesNotRestoreSessions(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->sends = [['result' => false, 'error' => 11]];
+        $transport->connectionResults = [false];
+        $client = $this->client($transport);
+        $this->seedState($client);
+
+        try {
+            $client->send('complete-message');
+            $this->fail('Expected reconnect failure');
+        } catch (Exception $exception) {
+            $this->assertStringContainsString('Failed to connect to MongoDB', $exception->getMessage());
+        }
+
+        $this->assertSame([[true], [true]], $transport->closes);
+        $this->assertStateCleared($client);
+    }
+
+    public function testImpossibleFrameLengthHardClosesTransport(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [['result' => pack('V', 20), 'error' => 0]];
+        $client = $this->client($transport);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertStringContainsString('Invalid response length', $exception->getMessage());
+        $this->assertSame([[true]], $transport->closes);
+    }
+
+    public function testOverlongFrameHardClosesTransport(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [['result' => pack('V*', 21, 1, 0, 2013, 0) . "\0x", 'error' => 0]];
+        $client = $this->client($transport);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertStringContainsString('Response length mismatch', $exception->getMessage());
+        $this->assertSame([[true]], $transport->closes);
+    }
+
+    public function testInvalidBsonHardClosesCoroutineTransport(): void
+    {
+        $transport = new CoroutineTransportDouble();
+        $transport->receives = [['result' => pack('V*', 26, 1, 0, 2013, 0) . "\0abcde", 'error' => 0]];
+        $client = $this->client($transport);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertStringContainsString('BSON', $exception->getMessage());
+        $this->assertSame(1, $transport->closes);
+    }
+
+    public function testOpReplyIsRejectedByOpMsgParser(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [[
+            'result' => pack('V*', 36, 1, 0, 1, 0, 0, 0, 0, 0),
+            'error' => 0,
+        ]];
+        $client = $this->client($transport);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertStringContainsString('Invalid response operation code: 1', $exception->getMessage());
+        $this->assertSame([[true]], $transport->closes);
+    }
+
+    public function testDecodedMongoCommandErrorDoesNotCloseTransport(): void
+    {
+        $transport = new SyncTransportDouble();
+        $transport->receives = [[
+            'result' => $this->frame([
+                'ok' => 0.0,
+                'errmsg' => 'duplicate key',
+                'code' => 11000,
+                'codeName' => 'DuplicateKey',
+            ]),
+            'error' => 0,
+        ]];
+        $client = $this->client($transport);
+
+        $exception = $this->receiveException($client);
+
+        $this->assertSame(11000, $exception->getCode());
+        $this->assertSame([], $transport->closes);
+        $this->assertTrue($transport->open);
+    }
+
+    private function assertStateCleared(Client $client): void
+    {
+        $this->assertFalse($this->get($client, 'isConnected'));
+        $this->assertContextCleared($client);
+    }
+
+    private function assertContextCleared(Client $client): void
+    {
+        $this->assertSame([], $this->get($client, 'sessions'));
+        $this->assertConnectionContextCleared($client);
+    }
+
+    private function assertConnectionContextCleared(Client $client): void
+    {
+        $this->assertNull($this->get($client, 'clusterTime'));
+        $this->assertNull($this->get($client, 'operationTime'));
+        $this->assertNull($this->get($client, 'replicaSet'));
+    }
+
+    private function client(SwooleClient|CoroutineClient $transport, float $timeout = 0.05): Client
+    {
+        $client = new Client('testing', 'mongo', 27017, 'root', 'example', timeout: $timeout);
+        $this->set($client, 'client', $transport);
+
+        return $client;
+    }
+
+    private function frame(array $document): string
+    {
+        $bson = (string) Document::fromPHP($document);
+
+        return pack('V*', 21 + strlen($bson), 1, 0, 2013, 0) . "\0" . $bson;
+    }
+
+    private function get(Client $client, string $property): mixed
+    {
+        $reflection = new ReflectionProperty(Client::class, $property);
+        $reflection->setAccessible(true);
+
+        return $reflection->getValue($client);
+    }
+
+    private function receive(Client $client): mixed
+    {
+        $reflection = new ReflectionMethod(Client::class, 'receive');
+        $reflection->setAccessible(true);
+
+        return $reflection->invoke($client);
+    }
+
+    private function receiveException(Client $client): Exception
+    {
+        try {
+            $this->receive($client);
+            $this->fail('Expected receive failure');
+        } catch (Exception $exception) {
+            return $exception;
+        }
+    }
+
+    private function reconnectingClient(SwooleClient|CoroutineClient $transport): ReconnectingClient
+    {
+        $client = new ReconnectingClient('testing', 'mongo', 27017, 'root', 'example');
+        $client->transport = $transport;
+        $this->set($client, 'client', $transport);
+
+        return $client;
+    }
+
+    private function seedState(Client $client): void
+    {
+        $this->seedConnectionContext($client);
+        $this->set($client, 'sessions', ['session' => ['id' => 'session']]);
+    }
+
+    private function seedConnectionContext(Client $client): void
+    {
+        $this->set($client, 'isConnected', true);
+        $this->set($client, 'clusterTime', (object) ['time' => 1]);
+        $this->set($client, 'operationTime', (object) ['time' => 2]);
+        $this->set($client, 'replicaSet', true);
+    }
+
+    private function set(Client $client, string $property, mixed $value): void
+    {
+        $reflection = new ReflectionProperty(Client::class, $property);
+        $reflection->setAccessible(true);
+        $reflection->setValue($client, $value);
+    }
+}