From 638591b58cda45e51ae41df6924fd211d008bc17 Mon Sep 17 00:00:00 2001
From: Jarvis Agent <jarvis@personal-assistant.ai>
Date: Tue, 10 Nov 2026 23:14:06 +0530
Subject: [PATCH 1/2] Fix timeout at the end of SDO block uploads

---
 canopen/sdo/client.py |  2 +-
 test/test_sdo.py      | 16 ++++++++++++++++
 2 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/canopen/sdo/client.py b/canopen/sdo/client.py
index 82ee0c88..7c936863 100644
--- a/canopen/sdo/client.py
+++ b/canopen/sdo/client.py
@@ -586,7 +586,7 @@ def _end_upload(self):
         try:
             response = self.sdo_client.read_response()
         except SdoCommunicationError:
-            self.abort(ABORT_TIMED_OUT)
+            self.sdo_client.abort(ABORT_TIMED_OUT)
             raise
         res_command, self._server_crc = struct.unpack_from("<BH", response)
         if res_command & 0xE0 != RESPONSE_BLOCK_UPLOAD:
diff --git a/test/test_sdo.py b/test/test_sdo.py
index ebf7a8f3..07a4a7a0 100644
--- a/test/test_sdo.py
+++ b/test/test_sdo.py
@@ -215,6 +215,22 @@ def test_block_upload(self):
             data = fp.read()
         self.assertEqual(data, 'Tiny Node - Mega Domains !')
 
+    def test_block_upload_timeout_waiting_for_end_response(self):
+        self.data = [
+            (TX, b'\xa4\x08\x10\x00\x7f\x00\x00\x00'),
+            (RX, b'\xc4\x08\x10\x00\x00\x00\x00\x00'),
+            (TX, b'\xa3\x00\x00\x00\x00\x00\x00\x00'),
+            (RX, b'\x81Tiny No'),
+            (TX, b'\xa2\x01\x7f\x00\x00\x00\x00\x00'),
+            # The server never sends the end-of-block upload response.
+            (TX, b'\x80\x00\x00\x00\x00\x00\x04\x05'),
+        ]
+        sdo = self.network[2].sdo[0x1008]
+        error = self.assertRaisesRegex(canopen.SdoCommunicationError, 'No SDO response received')
+        with sdo.open('rb', block_transfer=True) as fp, error:
+            fp.read()
+        self.assertEqual(self.data, [])
+
     def test_sdo_block_upload_retransmit(self):
         """Trigger a retransmit by only validating a block partially."""
         self.data = [

From 772648ab02cdbc1be33653fe2473305085217bc7 Mon Sep 17 00:00:00 2001
From: Dhairya Jangir <183471878+dhairyajangir@users.noreply.github.com>
Date: Thu, 12 Nov 2026 04:09:03 +0530
Subject: [PATCH 2/2] Include object addresses in client SDO aborts

Retain the initiating index/subindex in each stream and pass it through
all existing abort and retry-timeout paths. Encode the address without
inferring it from segment data or retaining global transfer state.

Keep direct abort() and abort(code) calls valid, document the optional
address keywords and subclass override boundary, and cover the wire
frames, retry behavior and original block-upload end timeout.

AI assisted with investigation, implementation, review and testing.
---
 canopen/sdo/client.py          | 169 +++++++++++++----
 doc/sdo.rst                    |  20 ++
 test/test_sdo.py               |   2 +-
 test/test_sdo_abort_address.py | 326 +++++++++++++++++++++++++++++++++
 4 files changed, 485 insertions(+), 32 deletions(-)
 create mode 100644 test/test_sdo_abort_address.py

diff --git a/canopen/sdo/client.py b/canopen/sdo/client.py
index 7c936863..5fce24ff 100644
--- a/canopen/sdo/client.py
+++ b/canopen/sdo/client.py
@@ -83,7 +83,12 @@ def read_response(self):
             raise SdoAbortedError(abort_code)
         return response
 
-    def request_response(self, sdo_request):
+    def request_response(self, sdo_request, *, index=0, subindex=0):
+        """Send a request, identifying its transfer in any timeout abort.
+
+        Stream operations supply the original object address. Low-level callers
+        can provide it explicitly; omitted values retain the zero-address default.
+        """
         retries_left = self.MAX_RETRIES
         if not self.responses.empty():
             # logger.warning("There were unexpected messages in the queue")
@@ -96,15 +101,25 @@ def request_response(self, sdo_request):
             except SdoCommunicationError as e:
                 retries_left -= 1
                 if not retries_left:
-                    self.abort(ABORT_TIMED_OUT)
+                    self.abort(ABORT_TIMED_OUT, index=index, subindex=subindex)
                     raise
                 logger.warning(str(e))
 
-    def abort(self, abort_code=ABORT_GENERAL_ERROR):
-        """Abort current transfer."""
+    def abort(self, abort_code=ABORT_GENERAL_ERROR, *, index=0, subindex=0):
+        """Abort a transfer with its object dictionary address.
+
+        :param int abort_code:
+            Reason for aborting. Defaults to a general error.
+        :param int index:
+            Object dictionary index of the transfer, or zero if unspecified.
+        :param int subindex:
+            Object dictionary subindex of the transfer, or zero if unspecified.
+
+        Stream operations supply their original address automatically. Existing
+        low-level calls without an address retain their zero-address behavior.
+        """
         request = bytearray(8)
-        request[0] = REQUEST_ABORTED
-        # TODO: Is it necessary to include index and subindex?
+        SDO_STRUCT.pack_into(request, 0, REQUEST_ABORTED, index, subindex)
         struct.pack_into("<L", request, 4, abort_code)
         self.send_request(request)
         logger.error("Transfer aborted by client with code 0x%08X", abort_code)
@@ -242,6 +257,8 @@ def __init__(self, sdo_client, index, subindex=0):
         :param int subindex:
             Object dictionary sub-index to read from.
         """
+        self._index = index
+        self._subindex = subindex
         self._done = False
         self.sdo_client = sdo_client
         self._toggle = 0
@@ -251,7 +268,11 @@ def __init__(self, sdo_client, index, subindex=0):
                      sdo_client.rx_cobid - 0x600)
         request = bytearray(8)
         SDO_STRUCT.pack_into(request, 0, REQUEST_UPLOAD, index, subindex)
-        response = sdo_client.request_response(request)
+        response = sdo_client.request_response(
+            request,
+            index=self._index,
+            subindex=self._subindex,
+        )
         res_command, res_index, res_subindex = SDO_STRUCT.unpack_from(response)
         res_data = response[4:8]
 
@@ -301,13 +322,25 @@ def read(self, size=-1):
         command |= self._toggle
         request = bytearray(8)
         request[0] = command
-        response = self.sdo_client.request_response(request)
+        response = self.sdo_client.request_response(
+            request,
+            index=self._index,
+            subindex=self._subindex,
+        )
         res_command, = struct.unpack_from("B", response)
         if res_command & 0xE0 != RESPONSE_SEGMENT_UPLOAD:
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError(f"Unexpected response 0x{res_command:02X}")
         if res_command & TOGGLE_BIT != self._toggle:
-            self.sdo_client.abort(ABORT_TOGGLE_NOT_ALTERNATED)
+            self.sdo_client.abort(
+                ABORT_TOGGLE_NOT_ALTERNATED,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError("Toggle bit mismatch")
         length = 7 - ((res_command >> 1) & 0x7)
         if res_command & NO_MORE_DATA:
@@ -348,6 +381,8 @@ def __init__(self, sdo_client, index, subindex=0, size=None, force_segment=False
         :param bool force_segment:
             Force use of segmented transfer regardless of size.
         """
+        self._index = index
+        self._subindex = subindex
         self.sdo_client = sdo_client
         self.size = size
         self.pos = 0
@@ -363,10 +398,18 @@ def __init__(self, sdo_client, index, subindex=0, size=None, force_segment=False
                 command |= SIZE_SPECIFIED
                 struct.pack_into("<L", request, 4, size)
             SDO_STRUCT.pack_into(request, 0, command, index, subindex)
-            response = sdo_client.request_response(request)
+            response = sdo_client.request_response(
+                request,
+                index=self._index,
+                subindex=self._subindex,
+            )
             res_command, = struct.unpack_from("B", response)
             if res_command != RESPONSE_DOWNLOAD:
-                self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+                self.sdo_client.abort(
+                    ABORT_INVALID_COMMAND_SPECIFIER,
+                    index=self._index,
+                    subindex=self._subindex,
+                )
                 raise SdoCommunicationError(
                     f"Unexpected response 0x{res_command:02X}")
         else:
@@ -392,10 +435,18 @@ def write(self, b):
                 raise AssertionError("More data received than expected")
             data = b.tobytes() if isinstance(b, memoryview) else b
             request = self._exp_header + data.ljust(4, b"\x00")
-            response = self.sdo_client.request_response(request)
+            response = self.sdo_client.request_response(
+                request,
+                index=self._index,
+                subindex=self._subindex,
+            )
             res_command, = struct.unpack_from("B", response)
             if res_command & 0xE0 != RESPONSE_DOWNLOAD:
-                self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+                self.sdo_client.abort(
+                    ABORT_INVALID_COMMAND_SPECIFIER,
+                    index=self._index,
+                    subindex=self._subindex,
+                )
                 raise SdoCommunicationError(
                     f"Unexpected response 0x{res_command:02X}")
             bytes_sent = len(b)
@@ -417,10 +468,18 @@ def write(self, b):
             command |= (7 - bytes_sent) << 1
             request[0] = command
             request[1:bytes_sent + 1] = b[0:bytes_sent]
-            response = self.sdo_client.request_response(request)
+            response = self.sdo_client.request_response(
+                request,
+                index=self._index,
+                subindex=self._subindex,
+            )
             res_command, = struct.unpack("B", response[0:1])
             if res_command & 0xE0 != RESPONSE_SEGMENT_DOWNLOAD:
-                self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+                self.sdo_client.abort(
+                    ABORT_INVALID_COMMAND_SPECIFIER,
+                    index=self._index,
+                    subindex=self._subindex,
+                )
                 raise SdoCommunicationError(
                     f"Unexpected response 0x{res_command:02X} "
                     f"(expected 0x{RESPONSE_SEGMENT_DOWNLOAD:02X})")
@@ -442,7 +501,11 @@ def close(self):
             command |= 7 << 1
             request = bytearray(8)
             request[0] = command
-            self.sdo_client.request_response(request)
+            self.sdo_client.request_response(
+                request,
+                index=self._index,
+                subindex=self._subindex,
+            )
             self._done = True
 
     def writable(self):
@@ -473,6 +536,8 @@ def __init__(self, sdo_client, index, subindex=0, request_crc_support=True):
         :param bool request_crc_support:
             If crc calculation should be requested when using block transfer
         """
+        self._index = index
+        self._subindex = subindex
         self._done = False
         self.sdo_client = sdo_client
         self.pos = 0
@@ -490,11 +555,19 @@ def __init__(self, sdo_client, index, subindex=0, request_crc_support=True):
             command |= CRC_SUPPORTED
         struct.pack_into("<BHBBB", request, 0,
                          command, index, subindex, self.blksize, 0)
-        response = sdo_client.request_response(request)
+        response = sdo_client.request_response(
+            request,
+            index=self._index,
+            subindex=self._subindex,
+        )
         res_command, res_index, res_subindex = SDO_STRUCT.unpack_from(response)
         if res_command & 0xE0 != RESPONSE_BLOCK_UPLOAD:
             self._error = True
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError(f"Unexpected response 0x{res_command:02X}")
         # Check that the message is for us
         if res_index != index or res_subindex != subindex:
@@ -551,7 +624,11 @@ def read(self, size=-1):
             if self._done:
                 if self._server_crc != self._crc.final():
                     self._error = True
-                    self.sdo_client.abort(ABORT_CRC_ERROR)
+                    self.sdo_client.abort(
+                        ABORT_CRC_ERROR,
+                        index=self._index,
+                        subindex=self._subindex,
+                    )
                     raise SdoCommunicationError("CRC is not OK")
                 logger.info("CRC is OK")
         self.pos += len(data)
@@ -571,7 +648,7 @@ def _retransmit(self):
                 self._ackseq = seqno
                 return response
         self._error = True
-        self.sdo_client.abort(ABORT_TIMED_OUT)
+        self.sdo_client.abort(ABORT_TIMED_OUT, index=self._index, subindex=self._subindex)
         raise SdoCommunicationError("Some data was lost and could not be retransmitted")
 
     def _ack_block(self):
@@ -586,16 +663,24 @@ def _end_upload(self):
         try:
             response = self.sdo_client.read_response()
         except SdoCommunicationError:
-            self.sdo_client.abort(ABORT_TIMED_OUT)
+            self.sdo_client.abort(ABORT_TIMED_OUT, index=self._index, subindex=self._subindex)
             raise
         res_command, self._server_crc = struct.unpack_from("<BH", response)
         if res_command & 0xE0 != RESPONSE_BLOCK_UPLOAD:
             self._error = True
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError(f"Unexpected response 0x{res_command:02X}")
         if res_command & 0x3 != END_BLOCK_TRANSFER:
             self._error = True
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError("Server did not end transfer as expected")
         # Return number of bytes not used in last message
         return (res_command >> 2) & 0x7
@@ -641,6 +726,8 @@ def __init__(self, sdo_client, index, subindex=0, size=None, request_crc_support
         :param bool request_crc_support:
             If crc calculation should be requested when using block transfer
         """
+        self._index = index
+        self._subindex = subindex
         self.sdo_client = sdo_client
         self.size = size
         self.pos = 0
@@ -662,15 +749,23 @@ def __init__(self, sdo_client, index, subindex=0, size=None, request_crc_support
         else:
             logger.warning("Data size has not been specified")
         SDO_STRUCT.pack_into(request, 0, command, index, subindex)
-        response = sdo_client.request_response(request)
+        response = sdo_client.request_response(
+            request,
+            index=self._index,
+            subindex=self._subindex,
+        )
         res_command, res_index, res_subindex = SDO_STRUCT.unpack_from(response)
         if res_command & 0xE0 != RESPONSE_BLOCK_DOWNLOAD:
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError(
                 f"Unexpected response 0x{res_command:02X}")
         # Check that the message is for us
         if res_index != index or res_subindex != subindex:
-            self.sdo_client.abort()
+            self.sdo_client.abort(index=self._index, subindex=self._subindex)
             raise SdoCommunicationError(
                 f"Node returned a value for {pretty_index(res_index, res_subindex)} instead, "
                 "maybe there is another SDO client communicating "
@@ -750,15 +845,23 @@ def _block_ack(self):
         try:
             response = self.sdo_client.read_response()
         except SdoCommunicationError:
-            self.sdo_client.abort(ABORT_TIMED_OUT)
+            self.sdo_client.abort(ABORT_TIMED_OUT, index=self._index, subindex=self._subindex)
             raise
         res_command, ackseq, blksize = struct.unpack_from("BBB", response)
         if res_command & 0xE0 != RESPONSE_BLOCK_DOWNLOAD:
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError(
                 f"Unexpected response 0x{res_command:02X}")
         if res_command & 0x3 != BLOCK_TRANSFER_RESPONSE:
-            self.sdo_client.abort(ABORT_INVALID_COMMAND_SPECIFIER)
+            self.sdo_client.abort(
+                ABORT_INVALID_COMMAND_SPECIFIER,
+                index=self._index,
+                subindex=self._subindex,
+            )
             raise SdoCommunicationError("Server did not respond with a "
                                         "block download response")
         if ackseq != self._blksize:
@@ -813,7 +916,11 @@ def close(self):
             # Add CRC
             struct.pack_into("<H", request, 1, self._crc.final())
         logger.debug("Ending block transfer...")
-        response = self.sdo_client.request_response(request)
+        response = self.sdo_client.request_response(
+            request,
+            index=self._index,
+            subindex=self._subindex,
+        )
         res_command, = struct.unpack_from("B", response)
         if not res_command & END_BLOCK_TRANSFER:
             raise SdoCommunicationError("Block download unsuccessful")
diff --git a/doc/sdo.rst b/doc/sdo.rst
index 29314028..d317f46d 100644
--- a/doc/sdo.rst
+++ b/doc/sdo.rst
@@ -103,6 +103,26 @@ server supports it. This is done through the file object interface::
    Block transfer is still in experimental stage!
 
 
+Client aborts
+-------------
+
+When a stream aborts a transfer, its abort frame includes the index and subindex
+used to open that stream. For a manually constructed transfer, these can also be
+passed explicitly::
+
+    node.sdo.abort(0x05040000, index=0x2345, subindex=0x67)
+
+Direct calls to ``abort()`` or ``abort(code)`` continue to send zero address
+fields. The same fallback applies to ``request_response(request)`` without
+explicit ``index`` and ``subindex`` keyword arguments; the client does not infer
+an address from arbitrary request payloads or remember the last stream's address.
+
+Custom ``SdoClient`` subclasses overriding ``abort`` or ``request_response`` must
+accept and forward the optional ``index`` and ``subindex`` keyword arguments used
+by the stream classes. Overrides accepting only the older positional arguments
+need to be updated before using those streams.
+
+
 API
 ---
 
diff --git a/test/test_sdo.py b/test/test_sdo.py
index 07a4a7a0..631612b9 100644
--- a/test/test_sdo.py
+++ b/test/test_sdo.py
@@ -223,7 +223,7 @@ def test_block_upload_timeout_waiting_for_end_response(self):
             (RX, b'\x81Tiny No'),
             (TX, b'\xa2\x01\x7f\x00\x00\x00\x00\x00'),
             # The server never sends the end-of-block upload response.
-            (TX, b'\x80\x00\x00\x00\x00\x00\x04\x05'),
+            (TX, b'\x80\x08\x10\x00\x00\x00\x04\x05'),
         ]
         sdo = self.network[2].sdo[0x1008]
         error = self.assertRaisesRegex(canopen.SdoCommunicationError, 'No SDO response received')
diff --git a/test/test_sdo_abort_address.py b/test/test_sdo_abort_address.py
new file mode 100644
index 00000000..c386632d
--- /dev/null
+++ b/test/test_sdo_abort_address.py
@@ -0,0 +1,326 @@
+"""Check that every client-side abort retains the initiating object's address."""
+
+import gc
+import struct
+import time
+
+import pytest
+
+import canopen
+
+
+INDEX = 0x2345
+SUBINDEX = 0x67
+ADDRESS = struct.pack('<HB', INDEX, SUBINDEX)
+PAYLOAD = bytes.fromhex('deadbeef012345')
+TIMEOUT = 0x05040000
+INVALID_COMMAND = 0x05040001
+TOGGLE = 0x05030000
+CRC = 0x05040004
+GENERAL = 0x08000000
+
+
+def tx(data, delay=0):
+    return ('tx', data, delay)
+
+
+def rx(data):
+    return ('rx', data, 0)
+
+
+def initial(command, tail=bytes(4), address=ADDRESS):
+    return bytes([command]) + address + tail
+
+
+def abort(code, index=INDEX, subindex=SUBINDEX):
+    return struct.pack('<BHBL', 0x80, index, subindex, code)
+
+
+def upload_start():
+    return [tx(initial(0x40)), rx(initial(0x41, struct.pack('<L', 7)))]
+
+
+def download_start(size=7):
+    command = 0x20 if size is None else 0x21
+    tail = bytes(4) if size is None else struct.pack('<L', size)
+    return [tx(initial(command, tail)), rx(initial(0x60))]
+
+
+def block_upload_start(crc=False, response=None):
+    command = 0xa4 if crc else 0xa0
+    answer = initial(0xc4 if crc else 0xc0) if response is None else response
+    return [tx(initial(command, bytes.fromhex('7f000000'))), rx(answer)]
+
+
+def block_download_start(response=None):
+    answer = initial(0xa0, bytes.fromhex('01000000')) if response is None else response
+    return [tx(initial(0xc2, struct.pack('<L', 7))), rx(answer)]
+
+
+def block_download_close():
+    # Existing close behavior is preserved even after a block-ACK error.
+    return [tx(bytes.fromhex('c100000000000000')), rx(bytes.fromhex('a100000000000000'))]
+
+
+@pytest.fixture
+def wire():
+    """Use the same scripted network transport as the existing SDO wire tests."""
+    networks = []
+
+    def make(frames):
+        script = list(frames)
+        network = canopen.Network()
+        networks.append(network)
+
+        def send_message(can_id, data, remote=False):
+            assert not remote
+            direction, expected, delay = script.pop(0)
+            assert direction == 'tx'
+            assert can_id == 0x602
+            assert bytes(data) == expected
+            if delay:
+                time.sleep(delay)
+            while script and script[0][0] == 'rx':
+                _, response, _ = script.pop(0)
+                network.notify(0x582, response, 0.0)
+
+        network.send_message = send_message
+        client = network.add_node(canopen.RemoteNode(2, canopen.ObjectDictionary())).sdo
+        client.RESPONSE_TIMEOUT = 0.005
+        return client, script
+
+    yield make
+    gc.collect()
+    for network in networks:
+        network.disconnect()
+
+
+def use_stream(client, options, action=None):
+    with client.open(INDEX, SUBINDEX, buffering=0, **options) as stream:
+        if action == 'read':
+            stream.read(7)
+        elif action == 'write':
+            stream.write(PAYLOAD)
+        elif action == 'expedited':
+            stream.write(PAYLOAD[:3])
+
+
+@pytest.mark.parametrize('path', [
+    'upload-init', 'download-init', 'block-upload-init', 'block-download-init',
+    'upload-segment', 'download-expedited', 'download-segment',
+    'download-close', 'block-download-close',
+])
+def test_timeout_context_at_all_request_response_paths(wire, path):
+    """Retry exhaustion must use the stream address, including payload and close requests."""
+    cleanup = []
+    if path == 'upload-init':
+        frames = [tx(initial(0x40))]
+        options, action = {'mode': 'rb'}, None
+    elif path == 'download-init':
+        frames = [tx(initial(0x21, struct.pack('<L', 7)))]
+        cleanup = [tx(bytes.fromhex('0f00000000000000')), rx(bytes.fromhex('2000000000000000'))]
+        options, action = {'mode': 'wb', 'size': 7}, None
+    elif path == 'block-upload-init':
+        frames = [tx(initial(0xa0, bytes.fromhex('7f000000')))]
+        options, action = {'mode': 'rb', 'block_transfer': True, 'request_crc_support': False}, None
+    elif path == 'block-download-init':
+        frames = [tx(initial(0xc2, struct.pack('<L', 7)))]
+        options, action = {'mode': 'wb', 'size': 7, 'block_transfer': True,
+                           'request_crc_support': False}, None
+    elif path == 'upload-segment':
+        frames = upload_start() + [tx(bytes.fromhex('6000000000000000'))]
+        options, action = {'mode': 'rb'}, 'read'
+    elif path == 'download-expedited':
+        frames = [tx(initial(0x27, PAYLOAD[:3] + b'\x00'))]
+        options, action = {'mode': 'wb', 'size': 3}, 'expedited'
+    elif path == 'download-segment':
+        frames = download_start() + [tx(b'\x01' + PAYLOAD)]
+        options, action = {'mode': 'wb', 'size': 7}, 'write'
+    elif path == 'download-close':
+        frames = download_start(None) + [tx(bytes.fromhex('0f00000000000000'))]
+        options, action = {'mode': 'wb'}, None
+    else:
+        frames = block_download_start() + [tx(b'\x81' + PAYLOAD),
+                  rx(bytes.fromhex('a201010000000000')), tx(bytes.fromhex('c100000000000000'))]
+        options, action = {'mode': 'wb', 'size': 7, 'block_transfer': True,
+                           'request_crc_support': False}, 'write'
+    client, script = wire(frames + [tx(abort(TIMEOUT))] + cleanup)
+    with pytest.raises(canopen.SdoCommunicationError, match='No SDO response received'):
+        use_stream(client, options, action)
+    gc.collect()
+    assert script == []
+
+
+ERROR_SITES = [
+    'upload-command', 'upload-toggle', 'download-init-command',
+    'download-expedited-command', 'download-segment-command',
+    'block-upload-init-command', 'block-upload-crc', 'block-upload-retransmit-timeout',
+    'block-upload-end-timeout', 'block-upload-end-command', 'block-upload-end-subcommand',
+    'block-download-init-command', 'block-download-init-address',
+    'block-download-ack-timeout', 'block-download-ack-command', 'block-download-ack-subcommand',
+]
+
+
+@pytest.mark.parametrize('site', ERROR_SITES)
+def test_stream_error_abort_uses_original_address(wire, site):
+    """Original object context survives wrong peer fields, segment data and block errors."""
+    cleanup = []
+    code = INVALID_COMMAND
+    if site in ('upload-command', 'upload-toggle'):
+        answer = b'\x60' + PAYLOAD if site == 'upload-command' else b'\x11' + PAYLOAD
+        frames = upload_start() + [tx(bytes.fromhex('6000000000000000')), rx(answer)]
+        options, action = {'mode': 'rb'}, 'read'
+        if site == 'upload-toggle':
+            code = TOGGLE
+    elif site == 'download-init-command':
+        frames = [tx(initial(0x21, struct.pack('<L', 7))), rx(initial(0x40))]
+        cleanup = [tx(bytes.fromhex('0f00000000000000')), rx(bytes.fromhex('2000000000000000'))]
+        options, action = {'mode': 'wb', 'size': 7}, None
+    elif site == 'download-expedited-command':
+        frames = [tx(initial(0x27, PAYLOAD[:3] + b'\x00')), rx(initial(0x40))]
+        options, action = {'mode': 'wb', 'size': 3}, 'expedited'
+    elif site == 'download-segment-command':
+        frames = download_start() + [tx(b'\x01' + PAYLOAD), rx(bytes.fromhex('603172b4cf098afe'))]
+        options, action = {'mode': 'wb', 'size': 7}, 'write'
+    elif site == 'block-upload-init-command':
+        frames = block_upload_start(response=initial(0x60))
+        options, action = {'mode': 'rb', 'block_transfer': True, 'request_crc_support': False}, None
+    elif site.startswith('block-upload-'):
+        crc = site == 'block-upload-crc'
+        frames = block_upload_start(crc) + [tx(bytes.fromhex('a300000000000000'))]
+        options, action = {'mode': 'rb', 'block_transfer': True, 'request_crc_support': crc}, 'read'
+        if site == 'block-upload-retransmit-timeout':
+            # Delay the ACK send beyond this retransmission window; no clock is mocked.
+            frames += [rx(b'\x02' + PAYLOAD), tx(bytes.fromhex('a2007f0000000000'), delay=0.02)]
+            code = TIMEOUT
+        else:
+            frames += [rx(b'\x81' + PAYLOAD), tx(bytes.fromhex('a2017f0000000000'))]
+            if site == 'block-upload-end-timeout':
+                code = TIMEOUT
+            elif site == 'block-upload-end-command':
+                frames += [rx(bytes.fromhex('6000000000000000'))]
+            elif site == 'block-upload-end-subcommand':
+                frames += [rx(bytes.fromhex('c000000000000000'))]
+            else:
+                frames += [rx(bytes.fromhex('c100000000000000'))]
+                code = CRC
+    elif site in ('block-download-init-command', 'block-download-init-address'):
+        response = initial(0x60) if site.endswith('command') else initial(
+            0xa0, bytes.fromhex('01000000'), address=bytes.fromhex('efbe91'))
+        frames = block_download_start(response)
+        options, action = {'mode': 'wb', 'size': 7, 'block_transfer': True,
+                           'request_crc_support': False}, None
+        if site.endswith('address'):
+            code = GENERAL
+    else:
+        frames = block_download_start() + [tx(b'\x81' + PAYLOAD)]
+        options, action = {'mode': 'wb', 'size': 7, 'block_transfer': True,
+                           'request_crc_support': False}, 'write'
+        cleanup = block_download_close()
+        if site == 'block-download-ack-timeout':
+            code = TIMEOUT
+        elif site == 'block-download-ack-command':
+            frames += [rx(bytes.fromhex('6001010000000000'))]
+        else:
+            frames += [rx(bytes.fromhex('a001010000000000'))]
+    client, script = wire(frames + [tx(abort(code))] + cleanup)
+    with pytest.raises(canopen.SdoCommunicationError):
+        use_stream(client, options, action)
+    gc.collect()
+    assert script == []
+
+
+@pytest.mark.parametrize('args,code', [((), GENERAL), ((TIMEOUT,), TIMEOUT)])
+def test_direct_abort_legacy_forms(wire, args, code):
+    client, script = wire([tx(abort(code, 0, 0))])
+    client.abort(*args)
+    assert script == []
+
+
+@pytest.mark.parametrize('index,subindex', [(0, 0), (1, 1), (0xffff, 0xff)])
+def test_direct_abort_explicit_address_bounds(wire, index, subindex):
+    client, script = wire([tx(abort(0xffffffff, index, subindex))])
+    client.abort(0xffffffff, index=index, subindex=subindex)
+    assert script == []
+
+
+@pytest.mark.parametrize('index,subindex', [
+    (-1, SUBINDEX), (0x10000, SUBINDEX), (INDEX, -1), (INDEX, 0x100),
+])
+def test_invalid_abort_address_is_rejected_before_sending(wire, index, subindex):
+    client, script = wire([])
+    with pytest.raises(struct.error):
+        client.abort(TIMEOUT, index=index, subindex=subindex)
+    assert script == []
+
+
+@pytest.mark.parametrize('explicit_context', [False, True])
+def test_direct_request_response_keeps_no_context_fallback(wire, explicit_context):
+    expected = abort(TIMEOUT) if explicit_context else abort(TIMEOUT, 0, 0)
+    client, script = wire([tx(initial(0x40)), tx(expected)])
+    keywords = {'index': INDEX, 'subindex': SUBINDEX} if explicit_context else {}
+    with pytest.raises(canopen.SdoCommunicationError, match='No SDO response received'):
+        client.request_response(initial(0x40), **keywords)
+    assert script == []
+
+
+@pytest.mark.parametrize('success_on_retry', [False, True])
+def test_retry_policy_preserves_count_and_abort(wire, success_on_retry):
+    frames = [tx(initial(0x40)), tx(initial(0x40))]
+    if success_on_retry:
+        frames += [rx(initial(0x43, PAYLOAD[:4]))]
+    else:
+        frames += [tx(abort(TIMEOUT))]
+    client, script = wire(frames)
+    client.MAX_RETRIES = 2
+    if success_on_retry:
+        assert client.upload(INDEX, SUBINDEX) == PAYLOAD[:4]
+    else:
+        with pytest.raises(canopen.SdoCommunicationError, match='No SDO response received'):
+            client.upload(INDEX, SUBINDEX)
+    assert script == []
+
+
+@pytest.mark.parametrize('kind', ['upload', 'block-upload', 'block-download'])
+def test_remote_abort_is_not_echoed(wire, kind):
+    if kind == 'upload':
+        request, options = initial(0x40), {'mode': 'rb'}
+    elif kind == 'block-upload':
+        request = initial(0xa0, bytes.fromhex('7f000000'))
+        options = {'mode': 'rb', 'block_transfer': True, 'request_crc_support': False}
+    else:
+        request = initial(0xc2, struct.pack('<L', 7))
+        options = {'mode': 'wb', 'size': 7, 'block_transfer': True, 'request_crc_support': False}
+    client, script = wire([tx(request), rx(abort(0x06020000))])
+    with pytest.raises(canopen.SdoAbortedError) as error:
+        use_stream(client, options)
+    assert error.value.code == 0x06020000
+    assert script == []
+
+
+@pytest.mark.parametrize('kind', ['upload', 'block-upload'])
+def test_existing_wrong_address_response_does_not_add_abort(wire, kind):
+    wrong_address = bytes.fromhex('efbe91')
+    if kind == 'upload':
+        frames = [tx(initial(0x40)), rx(initial(0x41, struct.pack('<L', 7), wrong_address))]
+        options = {'mode': 'rb'}
+    else:
+        frames = block_upload_start(response=initial(0xc0, address=wrong_address))
+        options = {'mode': 'rb', 'block_transfer': True, 'request_crc_support': False}
+    client, script = wire(frames)
+    with pytest.raises(canopen.SdoCommunicationError, match='Node returned a value'):
+        use_stream(client, options)
+    assert script == []
+
+
+def test_success_then_timeout_uses_new_stream_and_direct_abort_stays_unset(wire):
+    second = struct.pack('<HB', 0x4567, 0x89)
+    frames = [tx(initial(0x40)), rx(initial(0x43, PAYLOAD[:4])),
+              tx(b'\x40' + second + bytes(4)), tx(abort(TIMEOUT, 0x4567, 0x89)),
+              tx(abort(GENERAL, 0, 0))]
+    client, script = wire(frames)
+    assert client.upload(INDEX, SUBINDEX) == PAYLOAD[:4]
+    with pytest.raises(canopen.SdoCommunicationError, match='No SDO response received'):
+        client.upload(0x4567, 0x89)
+    client.abort()
+    assert script == []