diff --git a/canopen/sdo/client.py b/canopen/sdo/client.py
index 82ee0c88..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.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 ebf7a8f3..631612b9 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\x08\x10\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 = [
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 == []