gh-160309: Fix StreamReader.read() silently dropping incomplete final character

When a byte stream ends with an incomplete multibyte sequence (e.g. a
truncated UTF-8 character), StreamReader.read() would silently discard
those bytes: the read loop exited on empty newdata while bytebuffer still
held the fragment, and bytebuffer was never decoded.

Fix by detecting leftover bytes in bytebuffer when the underlying stream
is exhausted and decoding them immediately.  The codec's stateless decode
function (e.g. utf_8_decode with final=True) then applies the error
handler: 'strict' raises UnicodeDecodeError, 'replace' emits the
replacement character, and 'ignore' drops the bytes - matching the
behaviour of bytes.decode() and io.TextIOWrapper.

Author: saiyam-jain-dev

Lib/codecs.py

@@ -525,6 +525,19 @@ def read(self, size=-1, chars=-1, firstline=False):
             self.charbuffer += newchars
             # there was no data available
             if not newdata:
+                # The stream is exhausted.  If there are leftover bytes in
+                # bytebuffer, they form an incomplete character.  Decode them
+                # now so that the error handler is applied (strict raises,
+                # replace emits the replacement character, ignore drops them).
+                if self.bytebuffer:
+                    try:
+                        newchars, decodedbytes = self.decode(
+                            self.bytebuffer, self.errors, True)
+                    except TypeError:
+                        newchars, decodedbytes = self.decode(
+                            self.bytebuffer, self.errors)
+                    self.bytebuffer = self.bytebuffer[decodedbytes:]
+                    self.charbuffer += newchars
                 break
         if chars < 0:
             # Return everything we've got

Lib/test/test_codecs.py

@@ -2103,6 +2103,22 @@ def test_pickle(self):
                 with self.assertRaisesRegex(TypeError, 'StreamReader'):
                     pickle.dumps(f, proto)
 
+    def test_read_incomplete_character(self):
+        # gh-160309: Incomplete multibyte sequence at the end of stream
+        # must not be silently dropped.
+        stream = io.BytesIO(b'ab\xe2')
+        f = self.reader(stream, errors='strict')
+        with self.assertRaises(UnicodeDecodeError):
+            f.read()
+
+        stream = io.BytesIO(b'ab\xe2')
+        f = self.reader(stream, errors='replace')
+        self.assertEqual(f.read(), 'ab\ufffd')
+
+        stream = io.BytesIO(b'ab\xe2')
+        f = self.reader(stream, errors='ignore')
+        self.assertEqual(f.read(), 'ab')
+
 
 class StreamWriterTest(unittest.TestCase):