@@ -86,6 +86,10 @@ Header Schemas
    :members:
    :show-inheritance:
 
+.. autoclass:: pcapkit.protocols.schema.internet.igmpv3.GroupRecordListField
+   :members:
+   :show-inheritance:
+
 Data Models
 -----------
 
 
@@ -298,6 +298,7 @@ def make(self,
              sources: 'Optional[Iterable[IPv4Address | str | bytes | int]]' = None,
              reserved_1: 'int' = 0,
              reserved_2: 'bytes' = b'\x00\x00',
+             num_group_records: 'Optional[int]' = None,
              records: 'Optional[Iterable[dict[str, Any] | Data_GroupRecord]]' = None,
              payload: 'bytes | ProtocolBase | Schema' = b'',
              **kwargs: 'Any') -> 'Schema_IGMPv3':
@@ -347,6 +348,11 @@ def make(self,
                 (:rfc:`3376#section-4.2.1`); only meaningful for a Report.
             reserved_2: The reserved 16 bits before the group-record count
                 (:rfc:`3376#section-4.2.1`); only meaningful for a Report.
+            num_group_records: Number of group records, as transmitted on the
+                wire; the length of ``records`` when not given, so a capture cut
+                inside its record list rebuilds as captured; only meaningful for
+                a Report. A count above the length of ``records`` is written as
+                given, as on a Report the capture cut short.
             records: Group records (:rfc:`3376#section-4.2.4`), each either
                 a :class:`~pcapkit.protocols.data.internet.igmpv3.GroupRecord`
                 or a :obj:`dict` of the keyword arguments
@@ -360,7 +366,9 @@ def make(self,
 
         Raises:
             ProtocolError: If ``max_resp_time`` is not an :obj:`int` from 0
-                to 255, or ``type`` is neither a Query nor a Report.
+                to 255, ``num_group_records`` is not an :obj:`int` from the
+                length of ``records`` to 65535, or ``type`` is neither a Query
+                nor a Report.
 
         """
         type_value = self._make_index(type, type_default, namespace=type_namespace,
@@ -375,13 +383,19 @@ def make(self,
                 self._make_group_record(record) if not isinstance(record, dict) else
                 self._make_group_record(**record) for record in records
             ]
+            if num_group_records is None:
+                num_group_records = len(record_list)
+            elif isinstance(num_group_records, bool) or not isinstance(num_group_records, int) \
+                    or not len(record_list) <= num_group_records <= 0xFFFF:
+                raise ProtocolError(f'{self.alias}: invalid num_group_records: {num_group_records!r} '
+                                    f'for {len(record_list)} record(s)')
             schema = Schema_IGMPv3(
                 type=type_value,
                 body=Schema_Report(
                     reserved_1=reserved_1,
                     checksum=field,
                     reserved_2=reserved_2,
-                    num_group_records=len(record_list),
+                    num_group_records=num_group_records,
                     records=record_list,
                     payload=payload,
                 ),
@@ -451,6 +465,8 @@ def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]':  # type: ignore[ov
                 'reserved_1': report.reserved_1,
                 'checksum': report.checksum,
                 'reserved_2': report.reserved_2,
+                # NOTE: The count as parsed, so a Report cut inside its records rebuilds as captured.
+                'num_group_records': report.get('num_group_records'),
                 'records': report.records,
                 'payload': cls._make_payload(data),
             }
 
@@ -3,7 +3,7 @@
 """header schema for internet group management protocol version 3"""
 
 import io
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from pcapkit.const.igmp.record_type import RecordType as Enum_RecordType
 from pcapkit.const.igmp.type import Type as Enum_Type
@@ -19,7 +19,7 @@
 
 if TYPE_CHECKING:
     from ipaddress import IPv4Address
-    from typing import Any, Optional
+    from typing import IO, Any, Optional
 
     from typing_extensions import TypedDict
 
@@ -221,6 +221,46 @@ def __init__(self, record_type: 'Enum_RecordType | int', aux_data_len: 'int',
                      aux_data: 'bytes') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,multiple-statements
 
 
+class GroupRecordListField(ListField):
+    """Group record list that ends with the data.
+
+    :attr:`Report.records` is sized by the IGMP length, which exceeds the
+    octets present when the capture cuts the Report short. A record read past
+    them consumes nothing, which :meth:`ListField.unpack
+    <pcapkit.corekit.fields.collections.ListField.unpack>` refuses to loop on,
+    so the list is sized by the octets present instead: the record the data
+    ends inside is kept as far as it goes, and :meth:`Schema.unpack
+    <pcapkit.protocols.schema.schema.Schema.unpack>` records the short read
+    (:issue:`2650`).
+
+    """
+
+    def unpack(self, buffer: 'bytes | IO[bytes]', packet: 'dict[str, Any]') -> 'list[GroupRecord]':
+        """Unpack field value from :obj:`bytes`.
+
+        Args:
+            buffer: Field buffer.
+            packet: Packet data.
+
+        Returns:
+            Unpacked group records.
+
+        """
+        if isinstance(buffer, bytes):
+            present = len(buffer)
+        else:
+            start = buffer.tell()
+            present = buffer.seek(0, io.SEEK_END) - start
+            buffer.seek(start, io.SEEK_SET)
+
+        declared = self._length
+        self._length = min(declared, present)
+        try:
+            return cast('list[GroupRecord]', super().unpack(buffer, packet))
+        finally:
+            self._length = declared
+
+
 @schema_final
 class Report(Schema):
     """Header schema for an IGMPv3 Membership Report's body (:rfc:`3376#section-4.2`).
@@ -243,9 +283,10 @@ class Report(Schema):
     reserved_2: 'bytes' = BytesField(length=2, default=b'\x00\x00')
     #: Number of group records, as received on the wire.
     num_group_records: 'int' = UInt16Field()
-    #: Group records; see :meth:`post_process` for how this is kept to
-    #: exactly :attr:`num_group_records` entries.
-    records: 'list[GroupRecord]' = ListField(
+    #: Group records, read no further than the data goes (see
+    #: :class:`GroupRecordListField`); see :meth:`post_process` for how this is
+    #: kept to exactly :attr:`num_group_records` entries.
+    records: 'list[GroupRecord]' = GroupRecordListField(
         length=lambda pkt: pkt['__length__'],
         item_type=SchemaField(schema=GroupRecord),
     )
 
@@ -148,9 +148,8 @@ def test_the_checksum_covers_the_whole_ip_payload(self) -> None:
                                              f'0x{whole[2:4].hex()} expected')])
 
     def test_truncated_and_off(self) -> None:
-        # A Report of IGMPv3 cut short is Raw, its group records unparsed, so the
-        # IGMPv3 case is a Query.
-        for name in ('v2-report', 'v3-query'):
+        # An IGMPv3 Report cut short parses as far as it goes (#2650).
+        for name in ('v2-report', 'v3-query', 'v3-report'):
             packet = message(MESSAGES[name][0][:2], MESSAGES[name][0][4:] + b'XY')
             with self.subTest(name):
                 self.assertEqual(self.status(wire.ipv4(packet, proto=2)[:-1]), (self.Status.TRUNCATED, []))
 
@@ -1136,6 +1136,194 @@ def test_make_builds_report_from_dict_records(self) -> None:
         self.assertEqual(schema.body.records[1].num_sources, 1)
 
 
+#: A Report of two group records with source lists: the first record's header
+#: is octets 8-15 and its sources 16-23, the second's header 24-31 and its
+#: sources 32-43.
+IGMPV3_TWO_RECORD_REPORT = igmpv3_report((
+    igmpv3_group_record(1, '239.1.1.1', ('10.0.0.1', '10.0.0.2')),
+    igmpv3_group_record(2, '239.2.2.2', ('10.0.0.3', '10.0.0.4', '10.0.0.5')),
+))
+
+
+class IGMPv3TruncatedReportTests(unittest.TestCase):
+    """A Report the capture cuts inside its record list parses as far as its
+    octets go and records the cut, as a cut Query does, rather than falling
+    back to ``Raw`` (#2650). Each datagram's IPv4 header declares the whole
+    Report, so the IGMP layer is handed fewer octets than its length."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def parse_cut(self, packet: 'bytes', keep: 'int'):
+        """Parse ``packet`` inside an IPv4 datagram cut to ``keep`` octets of it."""
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            outer = IPv4(protocol=2, src='10.0.0.1', dst='224.0.0.22', payload=packet).data
+            cut = outer[:len(outer) - len(packet) + keep]
+            parsed = IPv4(cut)
+        return cut, parsed
+
+    def assert_rebuilds(self, cut: 'bytes', parsed) -> None:
+        """Both layers rebuild the octets captured, from ``info`` and its dict."""
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        igmp = parsed.payload
+        with warnings.catch_warnings():
+            # a cut layer warns that its length runs past the data, which is the point
+            warnings.simplefilter('ignore')
+            self.assertEqual(IGMPv3.from_data(igmp.info).data, igmp.data)
+            self.assertEqual(IGMPv3.from_data(igmp.info.to_dict()).data, igmp.data)
+            self.assertEqual(IPv4.from_data(parsed.info).data, cut)
+
+    def test_whole_report_is_not_marked(self) -> None:
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.protocols.misc.no_payload import NoPayload
+
+        packet = IGMPV3_TWO_RECORD_REPORT
+        cut, parsed = self.parse_cut(packet, len(packet))
+        igmp = parsed.payload
+
+        self.assertIsInstance(igmp, IGMPv3)
+        self.assertNotIn('__short_read__', igmp.info)
+        self.assertEqual(igmp.info.length, len(packet))
+        self.assertEqual(
+            [(str(record.multicast_address), tuple(str(src) for src in record.sources))
+             for record in igmp.info.records],
+            [('239.1.1.1', ('10.0.0.1', '10.0.0.2')),
+             ('239.2.2.2', ('10.0.0.3', '10.0.0.4', '10.0.0.5'))],
+        )
+        self.assertIsInstance(igmp.payload, NoPayload)
+        self.assert_rebuilds(cut, parsed)
+
+    def test_report_cut_inside_a_record_header(self) -> None:
+        """Cut after the second record's Number of Sources, before its
+        Multicast Address: the first record is whole, the second is kept as
+        far as it was read."""
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.protocols.misc.no_payload import NoPayload
+
+        packet = IGMPV3_TWO_RECORD_REPORT
+        cut, parsed = self.parse_cut(packet, 28)
+        igmp = parsed.payload
+
+        self.assertIsInstance(igmp, IGMPv3)
+        self.assertEqual(igmp.info['__short_read__'], ('body', 27))
+        self.assertEqual(igmp.info.length, 28)
+        self.assertEqual(igmp.data, packet[:28])
+        self.assertEqual(igmp.info.num_group_records, 2)
+        self.assertEqual(len(igmp.info.records), 2)
+        first, second = igmp.info.records
+        self.assertEqual(str(first.multicast_address), '239.1.1.1')
+        self.assertEqual(tuple(str(src) for src in first.sources), ('10.0.0.1', '10.0.0.2'))
+        self.assertEqual(second.record_type, 2)
+        self.assertEqual(second.num_sources, 3)
+        self.assertIsInstance(igmp.payload, NoPayload)
+        self.assert_rebuilds(cut, parsed)
+
+    def test_report_cut_inside_a_source_list(self) -> None:
+        """Cut inside the first record's second source: one record of the two
+        the count declares, which the rebuild still writes as two."""
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.protocols.misc.no_payload import NoPayload
+
+        packet = IGMPV3_TWO_RECORD_REPORT
+        cut, parsed = self.parse_cut(packet, 22)
+        igmp = parsed.payload
+
+        self.assertIsInstance(igmp, IGMPv3)
+        self.assertEqual(igmp.info['__short_read__'], ('body', 21))
+        self.assertEqual(igmp.info.length, 22)
+        self.assertEqual(igmp.info.num_group_records, 2)
+        self.assertEqual(len(igmp.info.records), 1)
+        record = igmp.info.records[0]
+        self.assertEqual(str(record.multicast_address), '239.1.1.1')
+        self.assertEqual(record.num_sources, 2)
+        self.assertEqual(str(record.sources[0]), '10.0.0.1')
+        self.assertIsInstance(igmp.payload, NoPayload)
+        self.assert_rebuilds(cut, parsed)
+
+    def test_cut_additional_data_goes_to_the_next_layer(self) -> None:
+        """Octets past the counted records still go to the payload when the
+        capture ends inside them (:rfc:`3376#section-4.2.11`)."""
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.protocols.misc.raw import Raw
+
+        packet = igmpv3_report((igmpv3_group_record(1, '224.0.1.1', ('10.0.0.1',)),)) + b'\xca\xfe\xba\xbe'
+        cut, parsed = self.parse_cut(packet, len(packet) - 2)
+        igmp = parsed.payload
+
+        self.assertIsInstance(igmp, IGMPv3)
+        self.assertEqual(len(igmp.info.records), 1)
+        self.assertIsInstance(igmp.payload, Raw)
+        self.assertEqual(igmp.payload.info.packet, b'\xca\xfe')
+        self.assertEqual(igmp.info.length, len(packet) - 4)
+        self.assert_rebuilds(cut, parsed)
+
+    def test_record_list_reads_bytes_and_a_live_stream_alike(self) -> None:
+        """The record list is sized by the octets present, from a buffer or a
+        stream part-way through, and keeps its declared length afterwards."""
+        from pcapkit.protocols.schema.internet.igmpv3 import Report
+
+        records = IGMPV3_TWO_RECORD_REPORT[8:28]
+        field = Report.__fields__['records']({'__length__': 36})
+        stream = io.BytesIO(b'\xde\xad' + records)
+        stream.seek(2)
+
+        for name, buffer in (('bytes', records), ('stream', stream)):
+            with self.subTest(buffer=name), warnings.catch_warnings():
+                # the record cut short warns that its length runs past the data
+                warnings.simplefilter('ignore')
+                parsed = field.unpack(buffer, {})
+                self.assertEqual([str(record.multicast_address) for record in parsed],
+                                 ['239.1.1.1', '0.0.0.0'])
+                self.assertEqual(parsed[1].num_sources, 3)
+                self.assertEqual(field.length, 36)
+
+
+class IGMPv3RecordCountTests(unittest.TestCase):
+    """``IGMPv3.make`` writes the ``num_group_records`` it is given only as an
+    :obj:`int` from the length of ``records`` to 65535; any other is refused
+    rather than written altered or losing records or payload (#2617, #2650)."""
+
+    RECORDS = (
+        {'record_type': 1, 'multicast_address': '239.1.1.1'},
+        {'record_type': 2, 'multicast_address': '239.2.2.2'},
+    )
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_a_count_the_records_or_the_field_cannot_carry_is_refused(self) -> None:
+        import re
+
+        from pcapkit.const.igmp.type import Type
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        proto = object.__new__(IGMPv3)
+        for count, payload in ((65536, b''), (-1, b''), (1, b'XYZW'), (0, b'XYZW'),
+                               (True, b''), (2.0, b''), ('2', b'')):
+            with self.subTest(count=count, payload=payload), self.assertRaisesRegex(
+                    ProtocolError, rf'^IGMPv3: invalid num_group_records: {re.escape(repr(count))} for 2 record'):
+                proto.make(type=Type.IGMPv3_Membership_Report, records=self.RECORDS,
+                           num_group_records=count, payload=payload)
+
+    def test_a_count_from_the_records_to_the_field_maximum_is_written(self) -> None:
+        from pcapkit.const.igmp.type import Type
+        from pcapkit.protocols.internet.igmpv3 import IGMPv3
+
+        records = igmpv3_group_record(1, '239.1.1.1') + igmpv3_group_record(2, '239.2.2.2')
+        for count, wire in ((None, 2), (2, 2), (3, 3), (0xFFFF, 0xFFFF)):
+            with self.subTest(count=count):
+                built = IGMPv3(type=Type.IGMPv3_Membership_Report, records=self.RECORDS,
+                               num_group_records=count, payload=b'XYZW').data
+                self.assertEqual(built[6:8], wire.to_bytes(2, 'big'))
+                self.assertEqual(built[8:], records + b'XYZW')
+
+
 class IGMPDispatchTests(unittest.TestCase):
     """The IP protocol 2 registry entry is the abstract base, which picks the
     version class per message (#2439, #2448)."""
 
@@ -73,6 +73,9 @@ def _ipv6(code: int, body: bytes) -> bytes:
     ('ospf', 'application.ospf:OSPF', {}, bytes.fromhex('020100180102030400000000abcd0000') + bytes(8)),
     ('ipx', 'internet.ipx:IPX', {}, bytes.fromhex('ffff001e0000000000000101010101010451000000000202020202020452')),
     ('ipv4', 'internet.ipv4:IPv4', {}, bytes.fromhex('450000141234000040fd7b83c0000201c6336401')),
+    ('igmpv3-report', 'internet.ipv4:IPv4', {}, bytes.fromhex(
+        '45000040123400000102bd710a000001e0000016' '2200c7e100000002'
+        '01000002ef0101010a0000010a000002' '02000003ef0202020a0000030a0000040a000005')),
     ('ipv6', IPV6, {}, _ipv6(59, b'')),
     ('ipv6-chain', IPV6, {}, bytes.fromhex(
         '600000000020004000000000000000000000000000000001000000000000000000000000000000023c'