asyncio: avoid sharing exception object between StreamReader and close waiter

StreamReaderProtocol.connection_lost() set the same exception object
on both the StreamReader's waiter and the Stream's _closed waiter.
Since gh-90082, Future stores the traceback at set_exception() time
and restores it with with_traceback() on every result() call, which
mutates the exception in place. Sharing one object between two
futures caused the second await (typically writer.wait_closed() in an
except block) to rewrite the traceback of the in-flight exception
being handled, erasing the real failure site (readexactly) and
replacing it with wait_closed frames.

Fix by copying the exception for the _closed waiter so each future
owns an independent object. Copy falls back to reconstructing via
type(exc)(*exc.args) when copy.copy fails.

Fixes #156278

Author: aryansk

Lib/asyncio/streams.py

@@ -3,6 +3,7 @@
     'open_connection', 'start_server')
 
 import collections
+import copy
 import socket
 import sys
 import warnings
@@ -280,7 +281,29 @@ def connection_lost(self, exc):
             if exc is None:
                 self._closed.set_result(None)
             else:
-                self._closed.set_exception(exc)
+                # Avoid sharing the same exception object between the
+                # reader future and the close waiter.  Future.result()
+                # restores the traceback with `with_traceback()`, which
+                # mutates the exception in place; sharing one object
+                # between two futures rewrites the traceback of the
+                # in-flight exception being handled (gh-156278).
+                try:
+                    exc_copy = copy.copy(exc)
+                except Exception:
+                    try:
+                        exc_copy = type(exc)(*exc.args)
+                        # Preserve context attributes where possible.
+                        if hasattr(exc, "__cause__"):
+                            exc_copy.__cause__ = exc.__cause__
+                        if hasattr(exc, "__context__"):
+                            exc_copy.__context__ = exc.__context__
+                        if hasattr(exc, "__suppress_context__"):
+                            exc_copy.__suppress_context__ = exc.__suppress_context__
+                        if exc.__traceback__ is not None:
+                            exc_copy = exc_copy.with_traceback(exc.__traceback__)
+                    except Exception:
+                        exc_copy = exc
+                self._closed.set_exception(exc_copy)
         super().connection_lost(exc)
         self._stream_reader_wr = None
         self._task = None