diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7a873da6..bbf95346 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@
 - Added support for numbered databases in Valkey clusters (#1696)
 ### Changed
 ### Fixed
+- Fixed interrupted pipelines leaving unread replies on reused connections; failed pipelines now discard remaining commands and reset execution state (#1309)
 
 ## v3.7.0 (2026-10-08)
 ### Added
diff --git a/src/Connection/RelayConnection.php b/src/Connection/RelayConnection.php
index 8e287b3c..5b8077f7 100644
--- a/src/Connection/RelayConnection.php
+++ b/src/Connection/RelayConnection.php
@@ -114,6 +114,8 @@ class RelayConnection extends AbstractConnection
         if ($this->client->isConnected()) {
             $this->client->close();
         }
+
+        parent::disconnect();
     }
 
     /**
diff --git a/src/Pipeline/Pipeline.php b/src/Pipeline/Pipeline.php
index ab3b946b..83f4fe38 100644
--- a/src/Pipeline/Pipeline.php
+++ b/src/Pipeline/Pipeline.php
@@ -12,7 +12,6 @@
 
 namespace Predis\Pipeline;
 
-use Exception;
 use InvalidArgumentException;
 use Predis\ClientContextInterface;
 use Predis\ClientException;
@@ -308,7 +307,22 @@ class Pipeline implements ClientContextInterface
     public function flushPipeline($send = true)
     {
         if ($send && !$this->pipeline->isEmpty()) {
-            $responses = $this->executePipeline($this->getConnection(), $this->pipeline);
+            $connection = $this->getConnection();
+
+            try {
+                $responses = $this->executePipeline($connection, $this->pipeline);
+            } catch (Throwable $exception) {
+                $this->pipeline = new SplQueue();
+
+                try {
+                    if ($connection->isConnected()) {
+                        $connection->disconnect();
+                    }
+                } finally {
+                    throw $exception;
+                }
+            }
+
             $this->responses = array_merge($this->responses, $responses);
         } else {
             $this->pipeline = new SplQueue();
@@ -339,7 +353,7 @@ class Pipeline implements ClientContextInterface
      * @param mixed $callable Optional callback for execution.
      *
      * @return array
-     * @throws Exception
+     * @throws Throwable
      * @throws InvalidArgumentException
      */
     public function execute($callable = null)
@@ -348,7 +362,6 @@ class Pipeline implements ClientContextInterface
             throw new InvalidArgumentException('The argument must be a callable object.');
         }
 
-        $exception = null;
         $this->setRunning(true);
 
         try {
@@ -357,14 +370,12 @@ class Pipeline implements ClientContextInterface
             }
 
             $this->flushPipeline();
-        } catch (Exception $exception) {
-            // NOOP
-        }
+        } catch (Throwable $exception) {
+            $this->pipeline = new SplQueue();
 
-        $this->setRunning(false);
-
-        if ($exception) {
             throw $exception;
+        } finally {
+            $this->setRunning(false);
         }
 
         return $this->responses;
diff --git a/tests/Predis/Connection/RelayConnectionTest.php b/tests/Predis/Connection/RelayConnectionTest.php
index 9edde479..814e72f8 100644
--- a/tests/Predis/Connection/RelayConnectionTest.php
+++ b/tests/Predis/Connection/RelayConnectionTest.php
@@ -13,6 +13,7 @@
 namespace Predis\Connection;
 
 use PHPUnit\Framework\MockObject\MockObject;
+use Predis\Client;
 use Predis\ClientException;
 use Predis\Command\RawCommand;
 use Predis\NotSupportedException;
@@ -455,6 +456,22 @@ class RelayConnectionTest extends PredisTestCase
     // ---- INTEGRATION TESTS --------------------------------------------- //
     // ******************************************************************** //
 
+    /**
+     * @group connected
+     */
+    public function testClientReconnectsAfterDisconnect(): void
+    {
+        $client = new Client($this->getParameters(), ['connections' => 'relay']);
+
+        $this->assertEquals('PONG', $client->ping());
+        $client->disconnect();
+        $this->assertFalse($client->isConnected());
+
+        $this->assertEquals('PONG', $client->ping());
+        $this->assertTrue($client->isConnected());
+        $client->disconnect();
+    }
+
     /**
      * @group connected
      */
diff --git a/tests/Predis/Pipeline/PipelineCleanupTest.php b/tests/Predis/Pipeline/PipelineCleanupTest.php
new file mode 100644
index 00000000..37b8cdea
--- /dev/null
+++ b/tests/Predis/Pipeline/PipelineCleanupTest.php
@@ -0,0 +1,179 @@
+<?php
+
+/*
+ * This file is part of the Predis package.
+ *
+ * (c) 2009-2020 Daniele Alessandri
+ * (c) 2021-2026 Till Krüss
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Predis\Pipeline;
+
+use Predis\Client;
+use Predis\Command\Redis\ECHO_;
+use Predis\Connection\Cluster\ClusterInterface;
+use Predis\Connection\NodeConnectionInterface;
+use Predis\Connection\Parameters;
+use PredisTestCase;
+use RuntimeException;
+use Throwable;
+use TypeError;
+
+class PipelineCleanupTest extends PredisTestCase
+{
+    /**
+     * @group disconnected
+     * @dataProvider providePipelineFailures
+     */
+    public function testDisconnectsAfterFailureAndDiscardsRemainingCommands(
+        string $connectionClass,
+        string $method,
+        Throwable $failure
+    ): void {
+        $connection = $this->getMockBuilder($connectionClass)->getMock();
+        $connection->method('getParameters')->willReturn(new Parameters());
+        $connection->method('isConnected')->willReturn(true);
+        $connection->expects($this->once())->method('disconnect');
+        $connection->expects($this->once())->method($method)->willThrowException($failure);
+
+        $pipeline = new Pipeline(new Client($connection));
+        $pipeline->echo('first');
+        $pipeline->exists('missing');
+
+        try {
+            $pipeline->execute();
+            $this->fail('Expected pipeline execution to fail.');
+        } catch (Throwable $exception) {
+            $this->assertSame($failure, $exception);
+        }
+
+        // Reusing the context must not replay commands already sent to Redis.
+        $this->assertSame([], $pipeline->execute());
+    }
+
+    public function providePipelineFailures(): array
+    {
+        $failures = [];
+
+        foreach ([RuntimeException::class, TypeError::class] as $failureClass) {
+            foreach (['write', 'readResponse'] as $method) {
+                $failures[] = [NodeConnectionInterface::class, $method, new $failureClass('Interrupted pipeline')];
+            }
+
+            $failures[] = [ClusterInterface::class, 'getConnectionByCommand', new $failureClass('Interrupted pipeline')];
+        }
+
+        return $failures;
+    }
+
+    /**
+     * @group disconnected
+     */
+    public function testPreservesOriginalFailureWhenDisconnectFails(): void
+    {
+        $failure = new RuntimeException('Interrupted pipeline');
+        $connection = $this->getMockBuilder(NodeConnectionInterface::class)->getMock();
+        $connection->method('getParameters')->willReturn(new Parameters());
+        $connection->method('isConnected')->willReturn(true);
+        $connection->method('write')->willThrowException($failure);
+        $connection->expects($this->once())->method('disconnect')
+            ->willThrowException(new RuntimeException('Disconnect failed'));
+
+        $pipeline = new Pipeline(new Client($connection));
+        $pipeline->ping();
+
+        try {
+            $pipeline->flushPipeline();
+            $this->fail('Expected pipeline execution to fail.');
+        } catch (Throwable $exception) {
+            $this->assertSame($failure, $exception);
+        }
+    }
+
+    /**
+     * @group disconnected
+     * @dataProvider provideCallbackFailures
+     */
+    public function testDiscardsQueuedCommandsAndResetsRunningStateAfterCallbackFailure(Throwable $failure): void
+    {
+        $connection = $this->getMockBuilder(NodeConnectionInterface::class)->getMock();
+        $connection->method('getParameters')->willReturn(new Parameters());
+        $connection->expects($this->never())->method('disconnect');
+        $connection->expects($this->never())->method('write');
+        $connection->expects($this->never())->method('readResponse');
+        $pipeline = new Pipeline(new Client($connection));
+
+        try {
+            $pipeline->execute(static function (Pipeline $pipe) use ($failure) {
+                $pipe->set('aborted-write', 'value');
+                throw $failure;
+            });
+            $this->fail('Expected callback to fail.');
+        } catch (Throwable $exception) {
+            $this->assertSame($failure, $exception);
+        }
+
+        $this->assertSame([], $pipeline->execute());
+    }
+
+    public function provideCallbackFailures(): array
+    {
+        return [
+            'PHP error' => [new TypeError('Callback failed')],
+            'exception' => [new RuntimeException('Callback failed')],
+        ];
+    }
+
+    /**
+     * @group connected
+     * @requiresRedisVersion >= 2.0.0
+     * @dataProvider providePipelineConnections
+     */
+    public function testClientReceivesCorrectRepliesAfterParserFailure(string $pipelineClass, bool $persistent): void
+    {
+        $parameters = $this->getParameters(['persistent' => $persistent]);
+        $client = new Client($parameters, ['connections' => 'default']);
+        $prefix = 'pipeline-cleanup:' . bin2hex(random_bytes(8)) . ':';
+        $hash = $prefix . 'hash';
+        $sorted = $prefix . 'sorted';
+        $client->hmset($hash, ['field' => 'value']);
+        $client->zadd($sorted, ['member' => 1]);
+        $failure = new TypeError('Response parser failed');
+
+        $command = $this->getMockBuilder(ECHO_::class)->onlyMethods(['parseResponse'])->getMock();
+        $command->setArguments(['first']);
+        $command->expects($this->once())->method('parseResponse')->willThrowException($failure);
+
+        $pipeline = new $pipelineClass($client);
+        $pipeline->executeCommand($command);
+        $pipeline->exists($prefix . 'missing');
+
+        try {
+            $pipeline->execute();
+            $this->fail('Expected response parsing to fail.');
+        } catch (Throwable $exception) {
+            $this->assertSame($failure, $exception);
+        }
+
+        $this->assertFalse($client->isConnected());
+        $this->assertSame(['field' => 'value'], $client->hgetall($hash));
+        $this->assertSame(['member' => '1'], $client->zrange($sorted, 0, -1, ['withscores' => true]));
+        $pipeline->echo('next');
+        $this->assertSame(['next'], $pipeline->execute());
+        $client->del([$hash, $sorted]);
+        $client->disconnect();
+    }
+
+    public function providePipelineConnections(): array
+    {
+        return [
+            [Pipeline::class, false],
+            [Pipeline::class, true],
+            [Atomic::class, false],
+            [Atomic::class, true],
+        ];
+    }
+}