diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md
index 7ebacae7524f..52f00add630f 100644
--- a/sdk/eventhub/azure-eventhub/CHANGELOG.md
+++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Bugs Fixed
 
 - Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. Also fixed the async transport to apply default TCP socket settings even when no custom `socket_settings` are provided. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394))
+- Fixed a bug in the pyAMQP transport where decoding an incoming performative whose trailing null fields were omitted by the sender (permitted by AMQP 1.0 section 1.4) raised `IndexError`/`TypeError`. The decoded field list is now padded to the performative's full field count so omitted trailing fields read back as their AMQP-defined default, including the compact `list0` encoding where every field is omitted. A field encoded as an explicit null but whose declared default is non-null (for example a `max_frame_size` set to null so the connection would compare `None < 512`) now also reads back as that default.
 
 ## 5.15.1 (2025-11-11)
 
diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py
index 32ef0ddd9c12..68e571a45c56 100644
--- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py
+++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_decode.py
@@ -24,6 +24,7 @@
 
 from . import described
 from .message import Message, Header, Properties
+from . import performatives
 
 if TYPE_CHECKING:
     from .message import MessageDict
@@ -416,6 +417,48 @@ def decode_payload(buffer: memoryview) -> Message:
     return Message(**message_properties)
 
 
+# The AMQP-defined default of each wire field of a performative, keyed by its
+# frame-type code and ordered as the fields appear on the wire. The AMQP 1.0
+# spec (section 1.4) lets a sender omit trailing fields whose value is the
+# default, so an incoming performative list can be shorter than the full field
+# count. Padding the decoded list back up to the full count with these defaults
+# keeps positional (frame[N]) access and namedtuple unpacking safe, and makes an
+# omitted field read back as its default rather than None. That distinction
+# matters: an Open that omits max_frame_size means 4294967295, and the connection
+# compares it numerically (frame[2] < 512), which would raise on None.
+# The transfer performative (code 20) carries a trailing payload that is not a
+# wire field, so its _definition uses a None sentinel for that slot, which is
+# excluded here.
+_PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = {
+    # _code and _definition are assigned onto the performative classes at import
+    # time (see performatives.py), so pylint cannot see them statically.
+    # pylint: disable=protected-access,no-member
+    performative._code: [field.default for field in performative._definition if field is not None]  # type: ignore
+    for performative in (
+        performatives.OpenFrame,
+        performatives.BeginFrame,
+        performatives.AttachFrame,
+        performatives.FlowFrame,
+        performatives.TransferFrame,
+        performatives.DispositionFrame,
+        performatives.DetachFrame,
+        performatives.EndFrame,
+        performatives.CloseFrame,
+        performatives.SASLMechanism,
+        performatives.SASLInit,
+        performatives.SASLChallenge,
+        performatives.SASLResponse,
+        performatives.SASLOutcome,
+    )
+}
+
+# The number of wire fields for each performative, derived from the defaults
+# above so the two stay in lockstep.
+_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = {
+    code: len(defaults) for code, defaults in _PERFORMATIVE_FIELD_DEFAULTS.items()
+}
+
+
 def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
     # Ignore the first two bytes, they will always be the constructors for
     # described type then ulong.
@@ -432,6 +475,12 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
                 f"AMQP frame field count {count} exceeds maximum {_MAX_COMPOUND_COUNT}"
             )
         buffer = data[12:]
+    elif compound_list_type == 0x45:
+        # list0 0x45: an empty list with no size or count bytes. A sender may
+        # encode a performative whose fields are all omitted this way, so treat
+        # it as zero fields and let the padding below fill in the nulls.
+        count = 0
+        buffer = data[4:]
     else:
         # list8 0xc0: data[4] is size, data[5] is count (1 byte, bounded at 255).
         count = data[5]
@@ -439,6 +488,25 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
     fields: List[Optional[memoryview]] = [None] * count
     for i in range(count):
         buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+    # A sender may omit trailing fields whose value is the default (AMQP 1.0
+    # section 1.4), so pad the decoded list back up to the performative's full
+    # field count with each omitted field's default before any positional access
+    # or unpacking.
+    field_defaults = _PERFORMATIVE_FIELD_DEFAULTS.get(frame_type)
+    if field_defaults is not None:
+        if count < len(field_defaults):
+            fields.extend(field_defaults[count:])
+        # Only trailing fields may be omitted, so a sender that sets a later
+        # field while wanting an earlier one's default must encode that earlier
+        # field as an explicit null. A decoded null for a field whose AMQP
+        # default is non-null therefore also means that default; normalize it so
+        # it reads back identically to the omitted case. For example, an Open
+        # that nulls max_frame_size must still compare as 4294967295, not None,
+        # when _incoming_open evaluates frame[2] < 512. Fields whose declared
+        # default is null are left as None.
+        for index, default in enumerate(field_defaults):
+            if default is not None and fields[index] is None:
+                fields[index] = default
     if frame_type == 20:
         fields.append(buffer)
     return frame_type, fields
diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py
index 4254715abffa..884021e5f16c 100644
--- a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py
+++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_decode.py
@@ -1,5 +1,14 @@
+import pathlib
 import pytest
-from azure.eventhub._pyamqp._decode import _decode_decimal128, _decode_described, _decode_array_small, _decode_array_large
+from azure.eventhub._pyamqp._decode import (
+    _decode_decimal128,
+    _decode_described,
+    _decode_array_small,
+    _decode_array_large,
+    decode_frame,
+    _PERFORMATIVE_FIELD_COUNT,
+)
+from azure.eventhub._pyamqp import performatives
 from decimal import Decimal
 
 
@@ -45,3 +54,215 @@ def test_array_of_described_large():
     for i in range(256):
         assert output[i] == [b'n', b'v']
         assert output[i].descriptor == 1335734831060
+
+
+def _list8_frame(code, count, encoded_fields, payload=b""):
+    # Build a described performative frame using a list8 (0xc0) body:
+    # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list8 (0xc0),
+    # size, count, then the encoded field bytes and any trailing payload.
+    header = bytes([0x00, 0x53, code, 0xC0, len(encoded_fields) + 1, count])
+    return memoryview(header + encoded_fields + payload)
+
+
+# A sender may omit trailing fields whose value is the default (AMQP 1.0 section
+# 1.4), so an incoming performative list can be shorter than the full field
+# count. The decoder must pad it back to the full count so positional access and
+# namedtuple unpacking stay safe and omitted fields read back as their default.
+def test_short_open_is_padded_to_full_field_count():
+    # Open with only container_id set ("x"), 1 field on the wire out of 10.
+    frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == performatives.OpenFrame._code
+    assert len(fields) == 10
+    # Unpacking and fixed-index access must not raise on the omitted fields.
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.container_id == b"x"
+    assert open_frame.properties is None
+    assert fields[9] is None
+
+
+def test_short_open_materializes_non_null_field_defaults():
+    # An Open that omits max_frame_size/channel_max means their AMQP defaults,
+    # not null. _connection._incoming_open reads them positionally and numerically
+    # (frame[2] < 512, frame[3]); padding with None would raise TypeError there.
+    frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78]))
+    _, fields = decode_frame(frame)
+    assert fields[2] == 4294967295  # max_frame_size default
+    assert fields[3] == 65535  # channel_max default
+    # Exercise the exact comparison _incoming_open performs; must not raise.
+    assert not fields[2] < 512
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.max_frame_size == 4294967295
+    assert open_frame.channel_max == 65535
+
+
+def test_open_with_explicit_null_field_uses_default():
+    # Only trailing fields may be omitted, so an Open that sets a later field
+    # (channel_max) while wanting the default max_frame_size must encode
+    # max_frame_size as an explicit null. That null must still read back as the
+    # 4294967295 default, or _incoming_open's frame[2] < 512 raises TypeError.
+    # container_id="x", hostname=null, max_frame_size=null, channel_max=100.
+    frame = _list8_frame(
+        performatives.OpenFrame._code, 4, bytes([0xA1, 0x01, 0x78, 0x40, 0x40, 0x52, 0x64])
+    )
+    _, fields = decode_frame(frame)
+    assert fields[2] == 4294967295  # explicit null normalized to the default
+    assert not fields[2] < 512  # the comparison _incoming_open performs
+    assert fields[3] == 100  # the explicitly set later field is preserved
+    assert fields[1] is None  # a null-default field (hostname) stays None
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.max_frame_size == 4294967295
+    assert open_frame.channel_max == 100
+
+
+def test_short_transfer_pads_fields_and_preserves_payload():
+    # Transfer with only handle (0) set, plus a message payload. The payload is
+    # appended after the fields and must survive the padding.
+    frame = _list8_frame(performatives.TransferFrame._code, 1, bytes([0x52, 0x00]), payload=b"\xde\xad")
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == performatives.TransferFrame._code
+    # 11 wire fields padded out, then the trailing payload appended (12 total).
+    assert len(fields) == 12
+    transfer = performatives.TransferFrame(*fields)
+    assert transfer.handle == 0
+    # Omitted boolean/uint fields read back as their AMQP defaults, not None.
+    assert transfer.message_format == 0
+    assert transfer.batchable is False
+    assert bytes(transfer.payload) == b"\xde\xad"
+
+
+def _list0_frame(code):
+    # Build a described performative whose body is an AMQP list0 (0x45): the most
+    # compact "all fields omitted" encoding, carrying no size or count bytes.
+    # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list0 (0x45).
+    return memoryview(bytes([0x00, 0x53, code, 0x45]))
+
+
+def _list32_frame(code, count, encoded_fields, payload=b""):
+    # Build a described performative frame using a list32 (0xd0) body, whose size
+    # and count are 4-byte big-endian. decode_frame ignores the size field and
+    # reads the count from data[8:12], so only the count must be accurate.
+    size = (len(encoded_fields) + 4).to_bytes(4, "big")
+    header = bytes([0x00, 0x53, code, 0xD0]) + size + count.to_bytes(4, "big")
+    return memoryview(header + encoded_fields + payload)
+
+
+# Begin/Attach/Disposition/Detach are namedtuples with no field defaults, so a
+# short positional unpack raised TypeError before the decoder padded to the full
+# field count. Only Open was covered above; exercise the rest of the no-default
+# performatives directly.
+@pytest.mark.parametrize(
+    "frame_cls",
+    [
+        performatives.AttachFrame,
+        performatives.BeginFrame,
+        performatives.DispositionFrame,
+        performatives.DetachFrame,
+    ],
+)
+def test_short_no_default_performative_is_padded(frame_cls):
+    # Each field's AMQP default, in wire order (the transfer payload sentinel
+    # is excluded, but these performatives have none).
+    defaults = [f.default for f in frame_cls._definition if f is not None]  # pylint: disable=protected-access
+    # A single null field on the wire, the rest omitted.
+    frame = _list8_frame(frame_cls._code, 1, bytes([0x40]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == frame_cls._code
+    assert len(fields) == len(defaults)
+    # The one wire field decoded as an explicit null; the omitted trailing fields
+    # are padded with their defaults (e.g. Disposition.batchable is False).
+    assert fields[0] is None
+    assert fields[1:] == defaults[1:]
+    # Namedtuple construction must not raise on the omitted (now padded) fields.
+    frame_cls(*fields)
+
+
+# The list32 (0xd0) body path is only taken for large frames and is otherwise
+# unexercised; confirm a short performative encoded as list32 pads identically to
+# the list8 case.
+def test_short_list32_is_padded_to_full_field_count():
+    frame = _list32_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == performatives.OpenFrame._code
+    assert len(fields) == 10
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.container_id == b"x"
+    assert fields[9] is None
+
+
+# SASLInit/SASLOutcome have required (no-default) fields, so a short SASL frame
+# crashed pre-fix. They are in the field-count map and must pad too.
+@pytest.mark.parametrize("frame_cls", [performatives.SASLInit, performatives.SASLOutcome])
+def test_short_sasl_frame_is_padded(frame_cls):
+    full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code]
+    frame = _list8_frame(frame_cls._code, 1, bytes([0x40]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == frame_cls._code
+    assert len(fields) == full_field_count
+    assert fields[-1] is None
+    frame_cls(*fields)
+
+
+# A performative with every field omitted may arrive as a list0 (0x45) body,
+# which has no count byte. The decoder must treat it as zero fields and pad up
+# to the full field count instead of indexing past the end of the buffer.
+@pytest.mark.parametrize("frame_cls", [performatives.EndFrame, performatives.CloseFrame])
+def test_list0_performative_pads_to_full_field_count(frame_cls):
+    frame = _list0_frame(frame_cls._code)
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == frame_cls._code
+    assert len(fields) == _PERFORMATIVE_FIELD_COUNT[frame_cls._code]
+    # Every omitted field reads back as None and unpacking must not raise.
+    performative = frame_cls(*fields)
+    assert performative.error is None
+
+
+@pytest.mark.parametrize(
+    "frame_cls,expected_count",
+    [
+        (performatives.OpenFrame, 10),
+        (performatives.BeginFrame, 8),
+        (performatives.AttachFrame, 14),
+        (performatives.FlowFrame, 11),
+        (performatives.TransferFrame, 11),
+        (performatives.DispositionFrame, 6),
+        (performatives.DetachFrame, 3),
+        (performatives.EndFrame, 1),
+        (performatives.CloseFrame, 1),
+    ],
+)
+def test_performative_field_count_matches_spec(frame_cls, expected_count):
+    # The padding target is the number of wire fields defined for each
+    # performative (the trailing transfer payload slot is excluded).
+    assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count
+
+
+# The _pyamqp engine is vendored identically into azure-eventhub and
+# azure-servicebus; a fix (like the padding above) must be applied to both
+# copies. Guard against the two copies silently drifting apart. The packages
+# ship separately, so skip when the sibling source is not present rather than
+# fail on a package-isolated checkout.
+_PYAMQP_COPIES = (
+    "sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp",
+    "sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp",
+)
+
+
+def _repo_root():
+    for parent in pathlib.Path(__file__).resolve().parents:
+        if (parent / "sdk").is_dir():
+            return parent
+    return None
+
+
+@pytest.mark.parametrize("filename", ["_decode.py", "_encode.py", "performatives.py"])
+def test_pyamqp_copies_are_byte_identical(filename):
+    root = _repo_root()
+    assert root is not None, "could not locate the repo root (no ancestor contains sdk/)"
+    eventhub_copy = root / _PYAMQP_COPIES[0] / filename
+    servicebus_copy = root / _PYAMQP_COPIES[1] / filename
+    if not (eventhub_copy.exists() and servicebus_copy.exists()):
+        pytest.skip("both _pyamqp copies are not present in this checkout")
+    assert (
+        eventhub_copy.read_bytes() == servicebus_copy.read_bytes()
+    ), f"{filename} has drifted between the eventhub and servicebus _pyamqp copies; apply to both."
diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md
index 2f38caeea96f..be9d3ecef980 100644
--- a/sdk/servicebus/azure-servicebus/CHANGELOG.md
+++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md
@@ -16,6 +16,7 @@
 - Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394))
 - Fixed a bug where passing a `fully_qualified_namespace` that included a port and/or trailing path (for example the `https://<namespace>.servicebus.windows.net:443/` form that Azure returns when provisioning a namespace) raised `ServiceBusAuthenticationError`. The namespace is now normalized to its bare host, matching the .NET and JavaScript SDKs. ([#44034](https://github.com/Azure/azure-sdk-for-python/issues/44034))
 - Fixed a bug where iterating over a `ServiceBusReceiver` suppressed automatic HTTP instrumentation (e.g. from `opentelemetry-instrumentation-httpx`/`requests`) while user code processed a received message, causing the user's own outbound HTTP spans to be dropped. The receive tracing span is now closed before the message is yielded to the caller, so suppression no longer leaks into message processing. ([#42755](https://github.com/Azure/azure-sdk-for-python/issues/42755))
+- Fixed a bug in the pyAMQP transport where decoding an incoming performative whose trailing null fields were omitted by the sender (permitted by AMQP 1.0 section 1.4) raised `IndexError`/`TypeError`. The decoded field list is now padded to the performative's full field count so omitted trailing fields read back as their AMQP-defined default, including the compact `list0` encoding where every field is omitted. A field encoded as an explicit null but whose declared default is non-null (for example a `max_frame_size` set to null so the connection would compare `None < 512`) now also reads back as that default.
 
 ### Other Changes
 
diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py
index 32ef0ddd9c12..68e571a45c56 100644
--- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py
+++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_decode.py
@@ -24,6 +24,7 @@
 
 from . import described
 from .message import Message, Header, Properties
+from . import performatives
 
 if TYPE_CHECKING:
     from .message import MessageDict
@@ -416,6 +417,48 @@ def decode_payload(buffer: memoryview) -> Message:
     return Message(**message_properties)
 
 
+# The AMQP-defined default of each wire field of a performative, keyed by its
+# frame-type code and ordered as the fields appear on the wire. The AMQP 1.0
+# spec (section 1.4) lets a sender omit trailing fields whose value is the
+# default, so an incoming performative list can be shorter than the full field
+# count. Padding the decoded list back up to the full count with these defaults
+# keeps positional (frame[N]) access and namedtuple unpacking safe, and makes an
+# omitted field read back as its default rather than None. That distinction
+# matters: an Open that omits max_frame_size means 4294967295, and the connection
+# compares it numerically (frame[2] < 512), which would raise on None.
+# The transfer performative (code 20) carries a trailing payload that is not a
+# wire field, so its _definition uses a None sentinel for that slot, which is
+# excluded here.
+_PERFORMATIVE_FIELD_DEFAULTS: Dict[int, List[Any]] = {
+    # _code and _definition are assigned onto the performative classes at import
+    # time (see performatives.py), so pylint cannot see them statically.
+    # pylint: disable=protected-access,no-member
+    performative._code: [field.default for field in performative._definition if field is not None]  # type: ignore
+    for performative in (
+        performatives.OpenFrame,
+        performatives.BeginFrame,
+        performatives.AttachFrame,
+        performatives.FlowFrame,
+        performatives.TransferFrame,
+        performatives.DispositionFrame,
+        performatives.DetachFrame,
+        performatives.EndFrame,
+        performatives.CloseFrame,
+        performatives.SASLMechanism,
+        performatives.SASLInit,
+        performatives.SASLChallenge,
+        performatives.SASLResponse,
+        performatives.SASLOutcome,
+    )
+}
+
+# The number of wire fields for each performative, derived from the defaults
+# above so the two stay in lockstep.
+_PERFORMATIVE_FIELD_COUNT: Dict[int, int] = {
+    code: len(defaults) for code, defaults in _PERFORMATIVE_FIELD_DEFAULTS.items()
+}
+
+
 def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
     # Ignore the first two bytes, they will always be the constructors for
     # described type then ulong.
@@ -432,6 +475,12 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
                 f"AMQP frame field count {count} exceeds maximum {_MAX_COMPOUND_COUNT}"
             )
         buffer = data[12:]
+    elif compound_list_type == 0x45:
+        # list0 0x45: an empty list with no size or count bytes. A sender may
+        # encode a performative whose fields are all omitted this way, so treat
+        # it as zero fields and let the padding below fill in the nulls.
+        count = 0
+        buffer = data[4:]
     else:
         # list8 0xc0: data[4] is size, data[5] is count (1 byte, bounded at 255).
         count = data[5]
@@ -439,6 +488,25 @@ def decode_frame(data: memoryview) -> Tuple[int, List[Any]]:
     fields: List[Optional[memoryview]] = [None] * count
     for i in range(count):
         buffer, fields[i] = _DECODE_BY_CONSTRUCTOR[buffer[0]](buffer[1:])
+    # A sender may omit trailing fields whose value is the default (AMQP 1.0
+    # section 1.4), so pad the decoded list back up to the performative's full
+    # field count with each omitted field's default before any positional access
+    # or unpacking.
+    field_defaults = _PERFORMATIVE_FIELD_DEFAULTS.get(frame_type)
+    if field_defaults is not None:
+        if count < len(field_defaults):
+            fields.extend(field_defaults[count:])
+        # Only trailing fields may be omitted, so a sender that sets a later
+        # field while wanting an earlier one's default must encode that earlier
+        # field as an explicit null. A decoded null for a field whose AMQP
+        # default is non-null therefore also means that default; normalize it so
+        # it reads back identically to the omitted case. For example, an Open
+        # that nulls max_frame_size must still compare as 4294967295, not None,
+        # when _incoming_open evaluates frame[2] < 512. Fields whose declared
+        # default is null are left as None.
+        for index, default in enumerate(field_defaults):
+            if default is not None and fields[index] is None:
+                fields[index] = default
     if frame_type == 20:
         fields.append(buffer)
     return frame_type, fields
diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py
new file mode 100644
index 000000000000..b22f6c7238be
--- /dev/null
+++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_pyamqp_decode.py
@@ -0,0 +1,216 @@
+import pathlib
+import pytest
+from azure.servicebus._pyamqp._decode import decode_frame, _PERFORMATIVE_FIELD_COUNT
+from azure.servicebus._pyamqp import performatives
+
+
+def _list8_frame(code, count, encoded_fields, payload=b""):
+    # Build a described performative frame using a list8 (0xc0) body:
+    # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list8 (0xc0),
+    # size, count, then the encoded field bytes and any trailing payload.
+    header = bytes([0x00, 0x53, code, 0xC0, len(encoded_fields) + 1, count])
+    return memoryview(header + encoded_fields + payload)
+
+
+# A sender may omit trailing fields whose value is the default (AMQP 1.0 section
+# 1.4), so an incoming performative list can be shorter than the full field
+# count. The decoder must pad it back to the full count so positional access and
+# namedtuple unpacking stay safe and omitted fields read back as their default.
+def test_short_open_is_padded_to_full_field_count():
+    # Open with only container_id set ("x"), 1 field on the wire out of 10.
+    frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == performatives.OpenFrame._code
+    assert len(fields) == 10
+    # Unpacking and fixed-index access must not raise on the omitted fields.
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.container_id == b"x"
+    assert open_frame.properties is None
+    assert fields[9] is None
+
+
+def test_short_open_materializes_non_null_field_defaults():
+    # An Open that omits max_frame_size/channel_max means their AMQP defaults,
+    # not null. _connection._incoming_open reads them positionally and numerically
+    # (frame[2] < 512, frame[3]); padding with None would raise TypeError there.
+    frame = _list8_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78]))
+    _, fields = decode_frame(frame)
+    assert fields[2] == 4294967295  # max_frame_size default
+    assert fields[3] == 65535  # channel_max default
+    # Exercise the exact comparison _incoming_open performs; must not raise.
+    assert not fields[2] < 512
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.max_frame_size == 4294967295
+    assert open_frame.channel_max == 65535
+
+
+def test_open_with_explicit_null_field_uses_default():
+    # Only trailing fields may be omitted, so an Open that sets a later field
+    # (channel_max) while wanting the default max_frame_size must encode
+    # max_frame_size as an explicit null. That null must still read back as the
+    # 4294967295 default, or _incoming_open's frame[2] < 512 raises TypeError.
+    # container_id="x", hostname=null, max_frame_size=null, channel_max=100.
+    frame = _list8_frame(
+        performatives.OpenFrame._code, 4, bytes([0xA1, 0x01, 0x78, 0x40, 0x40, 0x52, 0x64])
+    )
+    _, fields = decode_frame(frame)
+    assert fields[2] == 4294967295  # explicit null normalized to the default
+    assert not fields[2] < 512  # the comparison _incoming_open performs
+    assert fields[3] == 100  # the explicitly set later field is preserved
+    assert fields[1] is None  # a null-default field (hostname) stays None
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.max_frame_size == 4294967295
+    assert open_frame.channel_max == 100
+
+
+def test_short_transfer_pads_fields_and_preserves_payload():
+    # Transfer with only handle (0) set, plus a message payload. The payload is
+    # appended after the fields and must survive the padding.
+    frame = _list8_frame(performatives.TransferFrame._code, 1, bytes([0x52, 0x00]), payload=b"\xde\xad")
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == performatives.TransferFrame._code
+    # 11 wire fields padded out, then the trailing payload appended (12 total).
+    assert len(fields) == 12
+    transfer = performatives.TransferFrame(*fields)
+    assert transfer.handle == 0
+    # Omitted boolean/uint fields read back as their AMQP defaults, not None.
+    assert transfer.message_format == 0
+    assert transfer.batchable is False
+    assert bytes(transfer.payload) == b"\xde\xad"
+
+
+def _list0_frame(code):
+    # Build a described performative whose body is an AMQP list0 (0x45): the most
+    # compact "all fields omitted" encoding, carrying no size or count bytes.
+    # described-type ctor (0x00), ulong ctor (0x53), descriptor code, list0 (0x45).
+    return memoryview(bytes([0x00, 0x53, code, 0x45]))
+
+
+def _list32_frame(code, count, encoded_fields, payload=b""):
+    # Build a described performative frame using a list32 (0xd0) body, whose size
+    # and count are 4-byte big-endian. decode_frame ignores the size field and
+    # reads the count from data[8:12], so only the count must be accurate.
+    size = (len(encoded_fields) + 4).to_bytes(4, "big")
+    header = bytes([0x00, 0x53, code, 0xD0]) + size + count.to_bytes(4, "big")
+    return memoryview(header + encoded_fields + payload)
+
+
+# Begin/Attach/Disposition/Detach are namedtuples with no field defaults, so a
+# short positional unpack raised TypeError before the decoder padded to the full
+# field count. Only Open was covered above; exercise the rest of the no-default
+# performatives directly.
+@pytest.mark.parametrize(
+    "frame_cls",
+    [
+        performatives.AttachFrame,
+        performatives.BeginFrame,
+        performatives.DispositionFrame,
+        performatives.DetachFrame,
+    ],
+)
+def test_short_no_default_performative_is_padded(frame_cls):
+    # Each field's AMQP default, in wire order (the transfer payload sentinel
+    # is excluded, but these performatives have none).
+    defaults = [f.default for f in frame_cls._definition if f is not None]  # pylint: disable=protected-access
+    # A single null field on the wire, the rest omitted.
+    frame = _list8_frame(frame_cls._code, 1, bytes([0x40]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == frame_cls._code
+    assert len(fields) == len(defaults)
+    # The one wire field decoded as an explicit null; the omitted trailing fields
+    # are padded with their defaults (e.g. Disposition.batchable is False).
+    assert fields[0] is None
+    assert fields[1:] == defaults[1:]
+    # Namedtuple construction must not raise on the omitted (now padded) fields.
+    frame_cls(*fields)
+
+
+# The list32 (0xd0) body path is only taken for large frames and is otherwise
+# unexercised; confirm a short performative encoded as list32 pads identically to
+# the list8 case.
+def test_short_list32_is_padded_to_full_field_count():
+    frame = _list32_frame(performatives.OpenFrame._code, 1, bytes([0xA1, 0x01, 0x78]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == performatives.OpenFrame._code
+    assert len(fields) == 10
+    open_frame = performatives.OpenFrame(*fields)
+    assert open_frame.container_id == b"x"
+    assert fields[9] is None
+
+
+# SASLInit/SASLOutcome have required (no-default) fields, so a short SASL frame
+# crashed pre-fix. They are in the field-count map and must pad too.
+@pytest.mark.parametrize("frame_cls", [performatives.SASLInit, performatives.SASLOutcome])
+def test_short_sasl_frame_is_padded(frame_cls):
+    full_field_count = _PERFORMATIVE_FIELD_COUNT[frame_cls._code]
+    frame = _list8_frame(frame_cls._code, 1, bytes([0x40]))
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == frame_cls._code
+    assert len(fields) == full_field_count
+    assert fields[-1] is None
+    frame_cls(*fields)
+
+
+# A performative with every field omitted may arrive as a list0 (0x45) body,
+# which has no count byte. The decoder must treat it as zero fields and pad up
+# to the full field count instead of indexing past the end of the buffer.
+@pytest.mark.parametrize("frame_cls", [performatives.EndFrame, performatives.CloseFrame])
+def test_list0_performative_pads_to_full_field_count(frame_cls):
+    frame = _list0_frame(frame_cls._code)
+    frame_type, fields = decode_frame(frame)
+    assert frame_type == frame_cls._code
+    assert len(fields) == _PERFORMATIVE_FIELD_COUNT[frame_cls._code]
+    # Every omitted field reads back as None and unpacking must not raise.
+    performative = frame_cls(*fields)
+    assert performative.error is None
+
+
+@pytest.mark.parametrize(
+    "frame_cls,expected_count",
+    [
+        (performatives.OpenFrame, 10),
+        (performatives.BeginFrame, 8),
+        (performatives.AttachFrame, 14),
+        (performatives.FlowFrame, 11),
+        (performatives.TransferFrame, 11),
+        (performatives.DispositionFrame, 6),
+        (performatives.DetachFrame, 3),
+        (performatives.EndFrame, 1),
+        (performatives.CloseFrame, 1),
+    ],
+)
+def test_performative_field_count_matches_spec(frame_cls, expected_count):
+    # The padding target is the number of wire fields defined for each
+    # performative (the trailing transfer payload slot is excluded).
+    assert _PERFORMATIVE_FIELD_COUNT[frame_cls._code] == expected_count
+
+
+# The _pyamqp engine is vendored identically into azure-eventhub and
+# azure-servicebus; a fix (like the padding above) must be applied to both
+# copies. Guard against the two copies silently drifting apart. The packages
+# ship separately, so skip when the sibling source is not present rather than
+# fail on a package-isolated checkout.
+_PYAMQP_COPIES = (
+    "sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp",
+    "sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp",
+)
+
+
+def _repo_root():
+    for parent in pathlib.Path(__file__).resolve().parents:
+        if (parent / "sdk").is_dir():
+            return parent
+    return None
+
+
+@pytest.mark.parametrize("filename", ["_decode.py", "_encode.py", "performatives.py"])
+def test_pyamqp_copies_are_byte_identical(filename):
+    root = _repo_root()
+    assert root is not None, "could not locate the repo root (no ancestor contains sdk/)"
+    eventhub_copy = root / _PYAMQP_COPIES[0] / filename
+    servicebus_copy = root / _PYAMQP_COPIES[1] / filename
+    if not (eventhub_copy.exists() and servicebus_copy.exists()):
+        pytest.skip("both _pyamqp copies are not present in this checkout")
+    assert (
+        eventhub_copy.read_bytes() == servicebus_copy.read_bytes()
+    ), f"{filename} has drifted between the eventhub and servicebus _pyamqp copies; apply to both."