diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 4d9d8c0bf7..80117c8bd9 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -15,9 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::Identifier;
 use crate::utils::topic_size::MaxTopicSize;
+use crate::{IdKind, Identifier};
 use crate::{IggyMessage, utils::byte_size::IggyByteSize};
+use std::mem::size_of;
 use std::sync::Arc;
 use strum::{EnumDiscriminants, FromRepr, IntoStaticStr};
 use thiserror::Error;
@@ -533,6 +534,152 @@ pub enum IggyError {
     IncompatibleProtocolVersion(u32, u32, u32) = 14003,
 }
 
+/// Trait for serializing/deserializing individual [`IggyError`] payload fields.
+trait ErrorPayloadField: Sized {
+    fn write(&self, buf: &mut Vec<u8>);
+    fn read(buf: &[u8], pos: usize) -> (Self, usize);
+}
+
+impl ErrorPayloadField for usize {
+    fn write(&self, buf: &mut Vec<u8>) {
+        buf.extend_from_slice(&(*self as u64).to_le_bytes());
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        let bytes: [u8; size_of::<u64>()] = buf[pos..pos + size_of::<u64>()]
+            .try_into()
+            .expect("invalid usize payload");
+        (u64::from_le_bytes(bytes) as usize, pos + size_of::<u64>())
+    }
+}
+
+impl ErrorPayloadField for u32 {
+    fn write(&self, buf: &mut Vec<u8>) {
+        buf.extend_from_slice(&self.to_le_bytes());
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        let bytes: [u8; size_of::<Self>()] = buf[pos..pos + size_of::<Self>()]
+            .try_into()
+            .expect("invalid u32 payload");
+        (Self::from_le_bytes(bytes), pos + size_of::<Self>())
+    }
+}
+
+impl ErrorPayloadField for u64 {
+    fn write(&self, buf: &mut Vec<u8>) {
+        buf.extend_from_slice(&self.to_le_bytes());
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        let bytes: [u8; size_of::<Self>()] = buf[pos..pos + size_of::<Self>()]
+            .try_into()
+            .expect("invalid u64 payload");
+        (Self::from_le_bytes(bytes), pos + size_of::<Self>())
+    }
+}
+
+impl ErrorPayloadField for u16 {
+    fn write(&self, buf: &mut Vec<u8>) {
+        buf.extend_from_slice(&self.to_le_bytes());
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        let bytes: [u8; size_of::<Self>()] = buf[pos..pos + size_of::<Self>()]
+            .try_into()
+            .expect("invalid u16 payload");
+        (Self::from_le_bytes(bytes), pos + size_of::<Self>())
+    }
+}
+
+impl ErrorPayloadField for u8 {
+    fn write(&self, buf: &mut Vec<u8>) {
+        buf.push(*self);
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        (buf[pos], pos + 1)
+    }
+}
+
+impl ErrorPayloadField for Identifier {
+    fn write(&self, buf: &mut Vec<u8>) {
+        buf.push(self.kind as u8);
+        buf.push(self.length);
+        buf.extend_from_slice(&self.value);
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        let kind = match buf[pos] {
+            0 => IdKind::Numeric,
+            _ => IdKind::String,
+        };
+        let length = buf[pos + 1] as usize;
+        let value = buf[pos + 2..pos + 2 + length].to_vec();
+        (
+            Identifier {
+                kind,
+                length: length as u8,
+                value,
+            },
+            pos + 2 + length,
+        )
+    }
+}
+
+impl ErrorPayloadField for String {
+    fn write(&self, buf: &mut Vec<u8>) {
+        let len = self.len() as u32;
+        buf.extend_from_slice(&len.to_le_bytes());
+        buf.extend_from_slice(self.as_bytes());
+    }
+    fn read(buf: &[u8], pos: usize) -> (Self, usize) {
+        let len = u32::from_le_bytes(
+            buf[pos..pos + size_of::<u32>()]
+                .try_into()
+                .expect("invalid string length payload"),
+        ) as usize;
+        let data_start = pos + size_of::<u32>();
+        let data = String::from_utf8_lossy(&buf[data_start..data_start + len]).into_owned();
+        (data, data_start + len)
+    }
+}
+
+macro_rules! implement_error_payload {
+    ($($variant:ident($($field:ident: $ty:ty),* $(,)?)),* $(,)?) => {
+        /// Serialize the fields of a data-bearing error variant into a binary
+        /// payload. Returns an empty vector for unit variants and unhandled
+        /// variants.
+        pub fn write_payload(&self) -> Vec<u8> {
+            let mut buf = Vec::new();
+            match self {
+                $(IggyError::$variant($($field),*) => {
+                    $(ErrorPayloadField::write($field, &mut buf);)*
+                })*
+                _ => {}
+            }
+            buf
+        }
+
+        /// Reconstruct an [`IggyError`] from a server-provided code and a
+        /// structured binary payload. Falls back to [`IggyError::from_code`]
+        /// when the payload is empty or the code maps to a unit variant or
+        /// an unrecognized variant.
+        pub fn from_code_with_payload(code: u32, payload: &[u8]) -> Self {
+            if payload.is_empty() {
+                return IggyError::from_code(code);
+            }
+            let Some(disc) = IggyErrorDiscriminants::from_repr(code) else {
+                return IggyError::from_code(code);
+            };
+            let mut cursor: usize = 0;
+            match disc {
+                $(IggyErrorDiscriminants::$variant => {
+                    $(let ($field, off) = <$ty as ErrorPayloadField>::read(payload, cursor);
+                      cursor = off;)*
+                    let _ = cursor;
+                    IggyError::$variant($($field),*)
+                })*
+                _ => IggyError::from_code(code),
+            }
+        }
+    };
+}
+
 impl IggyError {
     pub fn as_code(&self) -> u32 {
         // SAFETY: SdkError specifies #[repr(u32)] representation.
@@ -553,6 +700,108 @@ impl IggyError {
             .map(|discriminant| discriminant.into())
             .unwrap_or("unknown error code")
     }
+
+    implement_error_payload! {
+        PartitionNotFound(partition_id: usize, topic_id: Identifier, stream_id: Identifier),
+        StreamIdNotFound(stream_id: Identifier),
+        TopicIdNotFound(stream_id: Identifier, topic_id: Identifier),
+        InvalidOffset(offset: u64),
+        ClientNotFound(client_id: u32),
+        CannotDeleteUser(user_id: u32),
+        CannotChangePermissions(user_id: u32),
+        ConsumerOffsetNotFound(offset: usize),
+        NoPartitions(topic_id: Identifier, stream_id: Identifier),
+        TopicFull(topic_id: Identifier, stream_id: Identifier),
+        ConsumerGroupIdNotFound(group_id: Identifier, topic_id: Identifier),
+        ConsumerGroupMemberNotFound(client_id: u32, group_id: Identifier, topic_id: Identifier),
+        ConsumerGroupPartitionNotOwned(client_id: u32, partition_id: u32),
+        CannotDeletePartitionDirectory(partition_id: usize, stream_id: usize, topic_id: usize),
+        CannotCreateConsumerGroupInfo(id: usize, topic_id: Identifier, stream_id: Identifier),
+        CannotDeleteConsumerGroupInfo(id: usize, topic_id: Identifier, stream_id: Identifier),
+        ConsumerGroupNameNotFound(name: String, topic_id: Identifier),
+        ConsumerGroupNameAlreadyExists(name: String, topic_id: Identifier),
+        InvalidReservedField(value: u64),
+        // Single-field String variants.
+        InvalidVersion(info: String),
+        CannotCreateBaseDirectory(path: String),
+        CannotCreateRuntimeDirectory(path: String),
+        CannotRemoveRuntimeDirectory(path: String),
+        CannotCreateStateDirectory(path: String),
+        CannotOpenDatabase(path: String),
+        ResourceNotFound(key: String),
+        CannotCloseWebSocketConnection(reason: String),
+        HttpError(info: String),
+        InvalidApiUrl(url: String),
+        InvalidJwtAlgorithm(alg: String),
+        CannotFetchJwks(url: String),
+        CannotParseHeaderKind(info: String),
+        CannotCreateStreamsDirectory(path: String),
+        StreamNameNotFound(name: String),
+        StreamDirectoryNotFound(path: String),
+        StreamNameAlreadyExists(name: String),
+        TopicDirectoryNotFound(path: String),
+        CannotDeleteConsumerOffsetsDirectory(path: String),
+        CannotDeleteConsumerOffsetFile(path: String),
+        CannotCreateConsumerOffsetsDirectory(path: String),
+        CannotReadConsumerOffsets(path: String),
+        CannotOpenConsumerOffsetsFile(path: String),
+        CannotCreateSegmentLogFile(path: String),
+        CannotCreateSegmentIndexFile(path: String),
+        CannotCreateSegmentTimeIndexFile(path: String),
+        CommandLengthError(info: String),
+        CannotBindToSocket(info: String),
+        IoError(info: String),
+        // Remaining compatible data-bearing variants.
+        InvalidStateEntryChecksum(a: u64, b: u64, c: u64),
+        InvalidIpAddress(ip: String, info: String),
+        HttpResponseError(status: u16, body: String),
+        PersonalAccessTokenAlreadyExists(token: String, user_id: u32),
+        PersonalAccessTokensLimitReached(user_id: u32, limit: u32),
+        PersonalAccessTokenExpired(token: String, user_id: u32),
+        CannotCreateStream(s: u32),
+        CannotCreateStreamDirectory(stream_id: u32, path: String),
+        CannotCreateStreamInfo(s: u32),
+        CannotUpdateStreamInfo(s: u32),
+        CannotOpenStreamInfo(s: u32),
+        CannotReadStreamInfo(s: u32),
+        CannotDeleteStream(s: u32),
+        CannotDeleteStreamDirectory(s: u32),
+        CannotCreateTopicsDirectory(stream_id: Identifier, path: String),
+        CannotCreateTopicDirectory(stream_id: usize, topic_id: usize, path: String),
+        CannotCreateTopic(stream_id: u32, topic_id: u32),
+        CannotCreateTopicInfo(stream_id: u32, topic_id: u32),
+        CannotDeleteTopic(stream_id: u32, topic_id: u32),
+        CannotDeleteTopicDirectory(stream_id: u32, topic_id: u32, path: String),
+        CannotOpenTopicInfo(stream_id: u32, topic_id: u32),
+        CannotReadTopicInfo(stream_id: u32, topic_id: u32),
+        CannotReadTopics(stream_id: u32),
+        CannotUpdateTopicInfo(stream_id: u32, topic_id: u32),
+        CannotCreatePartition(partition_id: usize, stream_id: usize, topic_id: usize),
+        CannotCreatePartitionDirectory(partition_id: usize, stream_id: usize, topic_id: usize),
+        CannotCreatePartitionsDirectory(stream_id: usize, topic_id: usize),
+        CannotDeletePartition(partition_id: usize, stream_id: Identifier, topic_id: Identifier),
+        InvalidBatchChecksum(a: u64, b: u64, c: u64),
+        InvalidHeaderKind(kind: u8),
+        InvalidIndexesByteSize(size: u32),
+        InvalidIndexesCount(count: u32, max: u32),
+        InvalidMessageChecksum(a: u64, b: u64, c: u64),
+        InvalidMessageTimestampDelta(delta: u64),
+        InvalidMessagesSize(size: u32, max: u32),
+        InvalidSegmentSize(size: u64),
+        InvalidSegmentsCount(count: u32),
+        InvalidSession(session: u64),
+        MissingIndex(index: u32),
+        NonZeroOffset(offset: u64, partition_id: u32),
+        NonZeroTimestamp(timestamp: u64, partition_id: u32),
+        NotResolvedConsumer(consumer_id: Identifier),
+        SegmentClosed(offset: u64, partition_id: usize),
+        ShardNotFound(shard_id: usize, topic_id: usize, stream_id: usize),
+        TimestampOutOfRange(timestamp: u64),
+        TooSmallMessage(size: u32, min: u32),
+        TopicNameAlreadyExists(name: String, topic_id: Identifier),
+        TopicNameNotFound(name: String, topic_id: String),
+        UnknownReplicatedCommand(code: u32),
+    }
 }
 
 impl PartialEq for IggyError {
@@ -594,4 +843,95 @@ mod tests {
             IggyError::from_code_as_string(GROUP_NAME_ERROR_CODE)
         )
     }
+
+    #[test]
+    fn from_code_reconstructs_unit_variant_preserving_code() {
+        let err = IggyError::from_code(IggyError::Unauthorized.as_code());
+        assert_eq!(err.as_code(), IggyError::Unauthorized.as_code());
+        assert!(matches!(err, IggyError::Unauthorized));
+    }
+
+    #[test]
+    fn from_code_fills_data_bearing_variant_with_default_fields() {
+        let code: u32 = 3007; // PartitionNotFound
+        let err = IggyError::from_code(code);
+        assert_eq!(err.as_code(), code);
+        // from_repr uses Default for data fields: partition 0, zeroed identifiers.
+        assert!(matches!(err, IggyError::PartitionNotFound(..)));
+        let error_string = err.to_string();
+        assert!(error_string.contains("Partition with ID: 0"));
+    }
+
+    #[test]
+    fn payload_round_trip_reconstructs_partition_not_found_with_real_fields() {
+        let original = IggyError::PartitionNotFound(
+            5,
+            crate::Identifier::numeric(1).unwrap(),
+            crate::Identifier::numeric(2).unwrap(),
+        );
+        let payload = original.write_payload();
+        assert!(!payload.is_empty());
+        let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload);
+        assert_eq!(reconstructed.as_code(), original.as_code());
+        assert!(matches!(&reconstructed, IggyError::PartitionNotFound(
+            5,
+            topic_id,
+            stream_id
+        ) if topic_id.value == 1u32.to_le_bytes().to_vec()
+          && stream_id.value == 2u32.to_le_bytes().to_vec()));
+    }
+
+    #[test]
+    fn payload_round_trip_reconstructs_stream_id_not_found_with_real_fields() {
+        let original = IggyError::StreamIdNotFound(crate::Identifier::numeric(42).unwrap());
+        let payload = original.write_payload();
+        assert!(!payload.is_empty());
+        let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload);
+        assert_eq!(reconstructed.as_code(), original.as_code());
+        assert!(
+            matches!(&reconstructed, IggyError::StreamIdNotFound(id) if id.value == 42u32.to_le_bytes().to_vec())
+        );
+    }
+
+    #[test]
+    fn payload_round_trip_reconstructs_invalid_offset_with_real_fields() {
+        let original = IggyError::InvalidOffset(123456);
+        let payload = original.write_payload();
+        assert!(!payload.is_empty());
+        let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload);
+        assert_eq!(reconstructed.as_code(), original.as_code());
+        assert!(matches!(reconstructed, IggyError::InvalidOffset(123456)));
+    }
+
+    #[test]
+    fn payload_round_trip_reconstructs_client_not_found_with_real_fields() {
+        let original = IggyError::ClientNotFound(99);
+        let payload = original.write_payload();
+        assert!(!payload.is_empty());
+        let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload);
+        assert_eq!(reconstructed.as_code(), original.as_code());
+        assert!(matches!(reconstructed, IggyError::ClientNotFound(99)));
+    }
+
+    #[test]
+    fn payload_round_trip_unit_variant_produces_empty_payload() {
+        let original = IggyError::Unauthorized;
+        let payload = original.write_payload();
+        assert!(payload.is_empty());
+        let reconstructed = IggyError::from_code_with_payload(original.as_code(), &payload);
+        assert_eq!(reconstructed, original);
+    }
+
+    #[test]
+    fn from_code_with_payload_falls_back_to_from_code_when_payload_is_empty() {
+        let code = IggyError::Unauthenticated.as_code();
+        let err = IggyError::from_code_with_payload(code, &[]);
+        assert_eq!(err, IggyError::Unauthenticated);
+    }
+
+    #[test]
+    fn from_code_unknown_code_falls_back_to_error_variant() {
+        let err = IggyError::from_code(0xDEAD_BEEF);
+        assert_eq!(err, IggyError::Error);
+    }
 }
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 9a731f6082..2c39862b6b 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -354,13 +354,24 @@ impl QuicClient {
                     .map_err(|_| IggyError::InvalidNumberEncoding)?,
             );
             if status != 0 {
+                let error_length = u32::from_le_bytes(
+                    buffer[4..RESPONSE_INITIAL_BYTES_LENGTH]
+                        .try_into()
+                        .map_err(|_| IggyError::InvalidNumberEncoding)?,
+                );
+                let body = if error_length > 0 {
+                    let body_end = RESPONSE_INITIAL_BYTES_LENGTH + error_length as usize;
+                    &buffer[RESPONSE_INITIAL_BYTES_LENGTH..body_end.min(buffer.len())]
+                } else {
+                    &[]
+                };
                 error!(
                     "Received an invalid response with status: {} ({}).",
                     status,
                     IggyError::from_code_as_string(status)
                 );
 
-                return Err(IggyError::from_code(status));
+                return Err(IggyError::from_code_with_payload(status, body));
             }
 
             let length = u32::from_le_bytes(
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index f827b9f55b..f2e1f06e57 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -327,6 +327,11 @@ impl TcpClient {
         stream: &mut ConnectionStreamKind,
     ) -> Result<Bytes, IggyError> {
         if status != 0 {
+            let mut body = BytesMut::new();
+            if length > 0 {
+                body.put_bytes(0, length as usize);
+                stream.read(&mut body).await?;
+            }
             // TEMP: See https://github.com/apache/iggy/pull/604 for context.
             if status == IggyErrorDiscriminants::TopicNameAlreadyExists as u32
                 || status == IggyErrorDiscriminants::StreamNameAlreadyExists as u32
@@ -347,7 +352,7 @@ impl TcpClient {
                 );
             }
 
-            return Err(IggyError::from_code(status));
+            return Err(IggyError::from_code_with_payload(status, &body));
         }
 
         trace!("Status: OK. Response length: {}", length);
@@ -1334,6 +1339,45 @@ mod tests {
         assert!(tcp_client.is_err());
     }
 
+    #[cfg(not(feature = "vsr"))]
+    #[tokio::test]
+    async fn should_reconstruct_data_bearing_error_from_payload_when_body_is_non_empty() {
+        let code: u32 = 3007; // PartitionNotFound
+        // partition_id=5, topic_id=1 (numeric), stream_id=2 (numeric)
+        let mut payload = Vec::new();
+        payload.extend_from_slice(&5u64.to_le_bytes()); // partition_id as u64
+        payload.push(0); // IdKind::Numeric
+        payload.push(4); // length
+        payload.extend_from_slice(&1u32.to_le_bytes()); // topic_id = 1
+        payload.push(0); // IdKind::Numeric
+        payload.push(4); // length
+        payload.extend_from_slice(&2u32.to_le_bytes()); // stream_id = 2
+        let mut stream = make_dummy_stream(&payload).await;
+        let result = TcpClient::handle_response(code, payload.len() as u32, &mut stream).await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.as_code(), code);
+        assert!(matches!(&err, IggyError::PartitionNotFound(
+            5,
+            topic_id,
+            stream_id
+        ) if topic_id.value == 1u32.to_le_bytes().to_vec()
+          && stream_id.value == 2u32.to_le_bytes().to_vec()));
+    }
+
+    #[cfg(not(feature = "vsr"))]
+    #[tokio::test]
+    async fn should_fallback_to_default_error_when_error_body_is_empty() {
+        let mut stream = make_dummy_stream(&[]).await;
+        let code: u32 = 3007; // PartitionNotFound
+        let result = TcpClient::handle_response(code, 0, &mut stream).await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.as_code(), code);
+        // Fallback: old server sends no body, so from_code produces Default fields.
+        assert!(matches!(err, IggyError::PartitionNotFound(..)));
+    }
+
     #[cfg(not(feature = "vsr"))]
     #[tokio::test]
     async fn should_return_ok_when_status_is_zero() {
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index d1e9ab859c..0bb674de38 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -184,7 +184,12 @@ pub(crate) fn decode_response(response: Bytes) -> Result<Bytes, IggyError> {
                 return Err(IggyError::InvalidCommand);
             }
             if let Some(error) = read_reply_status(header_bytes) {
-                return Err(error);
+                let body = if total_size > HEADER_SIZE {
+                    &response[HEADER_SIZE..total_size.min(response.len())]
+                } else {
+                    &[]
+                };
+                return Err(IggyError::from_code_with_payload(error.as_code(), body));
             }
             let operation = read_operation(header_bytes)?;
             split_metadata_result(operation, response.slice(HEADER_SIZE..total_size))
@@ -214,7 +219,12 @@ pub(crate) fn decode_response_split(
                 return Err(IggyError::InvalidCommand);
             }
             if let Some(error) = read_reply_status(header_bytes) {
-                return Err(error);
+                let payload = if expected_body > 0 {
+                    &body[..expected_body.min(body.len())]
+                } else {
+                    &[]
+                };
+                return Err(IggyError::from_code_with_payload(error.as_code(), payload));
             }
             let operation = read_operation(header_bytes)?;
             split_metadata_result(operation, body.slice(..expected_body))
@@ -593,6 +603,59 @@ mod tests {
         assert!(matches!(result, Err(IggyError::Unauthorized)));
     }
 
+    #[test]
+    fn reply_with_nonzero_data_bearing_code_reconstructs_from_payload() {
+        let code =
+            IggyError::PartitionNotFound(5, Default::default(), Default::default()).as_code();
+        // partition_id=5, topic_id=1 (numeric), stream_id=2 (numeric)
+        let mut payload = Vec::new();
+        payload.extend_from_slice(&5u64.to_le_bytes());
+        payload.push(0); // IdKind::Numeric
+        payload.push(4);
+        payload.extend_from_slice(&1u32.to_le_bytes());
+        payload.push(0); // IdKind::Numeric
+        payload.push(4);
+        payload.extend_from_slice(&2u32.to_le_bytes());
+        let body = Bytes::from(payload);
+        let header = ReplyHeader {
+            command: Command2::Reply,
+            size: (HEADER_SIZE + body.len()) as u32,
+            status: code,
+            ..Default::default()
+        };
+        let mut buf = [0u8; HEADER_SIZE];
+        buf.copy_from_slice(bytemuck::bytes_of(&header));
+        let result = decode_response_split(&buf, body);
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.as_code(), code);
+        assert!(matches!(&err, IggyError::PartitionNotFound(
+            5,
+            topic_id,
+            stream_id
+        ) if topic_id.value == 1u32.to_le_bytes().to_vec()
+          && stream_id.value == 2u32.to_le_bytes().to_vec()));
+    }
+
+    #[test]
+    fn reply_with_nonzero_data_bearing_code_falls_back_when_body_is_empty() {
+        let code =
+            IggyError::PartitionNotFound(5, Default::default(), Default::default()).as_code();
+        let header = ReplyHeader {
+            command: Command2::Reply,
+            size: HEADER_SIZE as u32,
+            status: code,
+            ..Default::default()
+        };
+        let mut buf = [0u8; HEADER_SIZE];
+        buf.copy_from_slice(bytemuck::bytes_of(&header));
+        let result = decode_response_split(&buf, Bytes::new());
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.as_code(), code);
+        assert!(matches!(err, IggyError::PartitionNotFound(..)));
+    }
+
     #[test]
     fn reply_with_zero_status_passes_body_through() {
         // status 0 is the ok channel: a non-metadata reply returns its body.
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 537d68de22..e1d3464d3f 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -829,6 +829,11 @@ impl WebSocketClient {
             );
 
             if status != 0 {
+                let mut body = vec![];
+                if length > 0 {
+                    body.resize(length, 0);
+                    stream.read(&mut body).await?;
+                }
                 // TEMP: See https://github.com/apache/iggy/pull/604 for context.
                 if status == IggyErrorDiscriminants::TopicNameAlreadyExists as u32
                     || status == IggyErrorDiscriminants::StreamNameAlreadyExists as u32
@@ -849,7 +854,7 @@ impl WebSocketClient {
                     );
                 }
 
-                return Err(IggyError::from_code(status));
+                return Err(IggyError::from_code_with_payload(status, &body));
             }
 
             if length == 0 {
diff --git a/core/server/src/sender/mod.rs b/core/server/src/sender/mod.rs
index c6a0b4acaa..25b76f0c86 100644
--- a/core/server/src/sender/mod.rs
+++ b/core/server/src/sender/mod.rs
@@ -203,7 +203,8 @@ pub(crate) async fn send_error_response<T>(
 where
     T: AsyncReadExt + AsyncWriteExt + Unpin,
 {
-    send_response(stream, &error.as_code().to_le_bytes(), &[]).await
+    let payload = error.write_payload();
+    send_response(stream, &error.as_code().to_le_bytes(), &payload).await
 }
 
 pub(crate) async fn send_response<T>(
diff --git a/foreign/python/tests/test_message_operations.py b/foreign/python/tests/test_message_operations.py
index da57637c05..a346f3d1e0 100644
--- a/foreign/python/tests/test_message_operations.py
+++ b/foreign/python/tests/test_message_operations.py
@@ -395,8 +395,8 @@ async def test_poll_messages_with_invalid_partition_id_raises(
         with pytest.raises(
             RuntimeError,
             match=(
-                r"Partition with ID: 0 for topic with ID: 0 "
-                r"for stream with ID: 0 was not found\."
+                r"Partition with ID: 1 for topic with ID: \d+ "
+                r"for stream with ID: \d+ was not found\."
             ),
         ):
             await iggy_client.poll_messages(