From 21c233b8680aad68ab832c03f3e45711307e3c5b Mon Sep 17 00:00:00 2001
From: youknowdot <dot@youknowone.org>
Date: Thu, 15 Oct 2026 15:28:13 +0900
Subject: [PATCH 1/3] Align asyncio initialization and future state with Python
 3.15

Reject native Future and Task reinitialization, validate task contexts, and restore task registration after failed eager context entry. Generate matching asyncio and concurrent Future sources and tests with update_lib from CPython d220409a13b77de7a6c30abe16ea90db01528edc. Preserve the existing pending-task reinitialization regression.

Assisted-by: OpenAI Codex:model version unavailable
---
 Lib/asyncio/futures.py                        |  21 ++-
 Lib/asyncio/tasks.py                          |  43 ++++-
 Lib/concurrent/futures/_base.py               |  29 ++-
 Lib/test/test_asyncio/test_futures.py         |  89 +++++----
 Lib/test/test_asyncio/test_tasks.py           | 175 ++++++++++++++----
 .../test_concurrent_futures/test_future.py    |  57 ++++++
 crates/stdlib/src/_asyncio.rs                 |  48 ++++-
 .../stdlib_asyncio_reinitialization_315.py    |  75 ++++++++
 8 files changed, 452 insertions(+), 85 deletions(-)
 create mode 100644 extra_tests/snippets/stdlib_asyncio_reinitialization_315.py

diff --git a/Lib/asyncio/futures.py b/Lib/asyncio/futures.py
index e8a99556b18..11858a0274a 100644
--- a/Lib/asyncio/futures.py
+++ b/Lib/asyncio/futures.py
@@ -79,6 +79,10 @@ def __init__(self, *, loop=None):
         loop object used by the future. If it's not provided, the future uses
         the default event loop.
         """
+        if self._loop is not None:
+            raise RuntimeError(f"{self.__class__.__name__} object is already "
+                                "initialized")
+
         if loop is None:
             self._loop = events.get_event_loop()
         else:
@@ -351,22 +355,19 @@ def _set_concurrent_future_state(concurrent, source):
 def _copy_future_state(source, dest):
     """Internal helper to copy state from another Future.
 
-    The other Future may be a concurrent.futures.Future.
+    The other Future must be a concurrent.futures.Future.
     """
-    assert source.done()
     if dest.cancelled():
         return
     assert not dest.done()
-    if source.cancelled():
+    done, cancelled, result, exception = source._get_snapshot()
+    assert done
+    if cancelled:
         dest.cancel()
+    elif exception is not None:
+        dest.set_exception(_convert_future_exc(exception))
     else:
-        exception = source.exception()
-        if exception is not None:
-            dest.set_exception(_convert_future_exc(exception))
-        else:
-            result = source.result()
-            dest.set_result(result)
-
+        dest.set_result(result)
 
 def _chain_future(source, destination):
     """Chain two futures so that when one completes, so does the other.
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 14d3c1ceb58..bdc577319c2 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -103,11 +103,20 @@ def __init__(self, coro, *, loop=None, name=None, context=None,
         self._coro = coro
         if context is None:
             self._context = contextvars.copy_context()
+        elif not isinstance(context, contextvars.Context):
+            # gh-157301: the passed value must be a contextvars.Context
+            self._log_destroy_pending = False
+            raise TypeError('a contextvars.Context was expected, '
+                            f'got {type(context).__name__}')
         else:
             self._context = context
 
         if eager_start and self._loop.is_running():
-            self.__eager_start()
+            try:
+                self.__eager_start()
+            except:
+                self._log_destroy_pending = False
+                raise
         else:
             self._loop.call_soon(self.__step, context=self._context)
             _py_register_task(self)
@@ -440,15 +449,13 @@ def _release_waiter(waiter, *args):
 async def wait_for(fut, timeout):
     """Wait for the single Future or coroutine to complete, with timeout.
 
-    Coroutine will be wrapped in Task.
-
     Returns result of the Future or coroutine.  When a timeout occurs,
-    it cancels the task and raises TimeoutError.  To avoid the task
-    cancellation, wrap it in shield().
+    it cancels fut and raises TimeoutError.  To prevent fut from being
+    cancelled, wrap it in shield().
 
-    If the wait is cancelled, the task is also cancelled.
+    If the wait is cancelled, fut is also cancelled.
 
-    If the task suppresses the cancellation and returns a value instead,
+    If fut suppresses the cancellation and returns a value instead,
     that value is returned.
 
     This function is a coroutine.
@@ -524,6 +531,7 @@ def _on_completion(f):
             timeout_handle.cancel()
         for f in fs:
             f.remove_done_callback(_on_completion)
+            futures.future_discard_from_awaited_by(f, cur_task)
 
     done, pending = set(), set()
     for f in fs:
@@ -542,6 +550,10 @@ async def _cancel_and_wait(fut):
     cb = functools.partial(_release_waiter, waiter)
     fut.add_done_callback(cb)
 
+    # gh-157058: awaiting the waiter leaves no edge on fut, add it here
+    cur_task = current_task()
+    futures.future_add_to_awaited_by(fut, cur_task)
+
     try:
         fut.cancel()
         # We cannot wait on *fut* directly to make
@@ -549,6 +561,7 @@ async def _cancel_and_wait(fut):
         await waiter
     finally:
         fut.remove_done_callback(cb)
+        futures.future_discard_from_awaited_by(fut, cur_task)
 
 
 class _AsCompletedIterator:
@@ -563,9 +576,11 @@ def __init__(self, aws, timeout):
         self._timeout_handle = None
 
         loop = events.get_event_loop()
+        self._cur_task = current_task()
         todo = {ensure_future(aw, loop=loop) for aw in set(aws)}
         for f in todo:
             f.add_done_callback(self._handle_completion)
+            futures.future_add_to_awaited_by(f, self._cur_task)
         if todo and timeout is not None:
             self._timeout_handle = (
                 loop.call_later(timeout, self._handle_timeout)
@@ -596,6 +611,7 @@ def __next__(self):
     def _handle_timeout(self):
         for f in self._todo:
             f.remove_done_callback(self._handle_completion)
+            futures.future_discard_from_awaited_by(f, self._cur_task)
             self._done.put_nowait(None)  # Sentinel for _wait_for_one().
         self._todo.clear()  # Can't do todo.remove(f) in the loop.
 
@@ -603,6 +619,7 @@ def _handle_completion(self, f):
         if not self._todo:
             return  # _handle_timeout() was here first.
         self._todo.remove(f)
+        futures.future_discard_from_awaited_by(f, self._cur_task)
         self._done.put_nowait(f)
         if not self._todo and self._timeout_handle is not None:
             self._timeout_handle.cancel()
@@ -767,6 +784,11 @@ def cancel(self, msg=None):
         return ret
 
 
+def _discard_awaited_by(children, waiter, outer):
+    for fut in children:
+        futures.future_discard_from_awaited_by(fut, waiter)
+
+
 def gather(*coros_or_futures, return_exceptions=False):
     """Return a future aggregating results from the given coroutines/futures.
 
@@ -900,6 +922,10 @@ def _done_callback(fut, cur_task=cur_task):
         children.append(fut)
 
     outer = _GatheringFuture(children, loop=loop)
+    if cur_task is not None:
+        # gh-157213: a child outliving gather() must lose the awaited-by edge
+        outer.add_done_callback(
+            functools.partial(_discard_awaited_by, children, cur_task))
     # Run done callbacks after GatheringFuture created so any post-processing
     # can be performed at this point
     # optimization: in the special case that *all* futures finished eagerly,
@@ -996,6 +1022,9 @@ def _outer_done_callback(outer):
             # Keep only one callback to log on cancel
             inner.remove_done_callback(_log_on_exception)
             inner.add_done_callback(_log_on_exception)
+            if cur_task is not None:
+                inner.remove_done_callback(_clear_awaited_by_callback)
+                futures.future_discard_from_awaited_by(inner, cur_task)
 
     if cur_task is not None:
         inner.add_done_callback(_clear_awaited_by_callback)
diff --git a/Lib/concurrent/futures/_base.py b/Lib/concurrent/futures/_base.py
index 96156f47acc..43774066c5a 100644
--- a/Lib/concurrent/futures/_base.py
+++ b/Lib/concurrent/futures/_base.py
@@ -563,6 +563,33 @@ def set_exception(self, exception):
             self._condition.notify_all()
         self._invoke_callbacks()
 
+    def _get_snapshot(self):
+        """Get a snapshot of the future's current state.
+
+        This method atomically retrieves the state in one lock acquisition,
+        which is significantly faster than multiple method calls.
+
+        Returns:
+            Tuple of (done, cancelled, result, exception)
+            - done: True if the future is done (cancelled or finished)
+            - cancelled: True if the future was cancelled
+            - result: The result if available and not cancelled
+            - exception: The exception if available and not cancelled
+        """
+        # Fast path: check if already finished without lock
+        if self._state == FINISHED:
+            return True, False, self._result, self._exception
+
+        # Need lock for other states since they can change
+        with self._condition:
+            # We have to check the state again after acquiring the lock
+            # because it may have changed in the meantime.
+            if self._state == FINISHED:
+                return True, False, self._result, self._exception
+            if self._state in {CANCELLED, CANCELLED_AND_NOTIFIED}:
+                return True, True, None, None
+            return False, False, None, None
+
     __class_getitem__ = classmethod(types.GenericAlias)
 
 class Executor(object):
@@ -676,5 +703,5 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 class BrokenExecutor(RuntimeError):
     """
-    Raised when a executor has become non-functional after a severe failure.
+    Raised when an executor has become non-functional after a severe failure.
     """
diff --git a/Lib/test/test_asyncio/test_futures.py b/Lib/test/test_asyncio/test_futures.py
index 785258a2749..9385a65e528 100644
--- a/Lib/test/test_asyncio/test_futures.py
+++ b/Lib/test/test_asyncio/test_futures.py
@@ -413,7 +413,7 @@ def func_repr(func):
     def test_copy_state(self):
         from asyncio.futures import _copy_future_state
 
-        f = self._new_future(loop=self.loop)
+        f = concurrent.futures.Future()
         f.set_result(10)
 
         newf = self._new_future(loop=self.loop)
@@ -421,7 +421,7 @@ def test_copy_state(self):
         self.assertTrue(newf.done())
         self.assertEqual(newf.result(), 10)
 
-        f_exception = self._new_future(loop=self.loop)
+        f_exception = concurrent.futures.Future()
         f_exception.set_exception(RuntimeError())
 
         newf_exception = self._new_future(loop=self.loop)
@@ -429,7 +429,7 @@ def test_copy_state(self):
         self.assertTrue(newf_exception.done())
         self.assertRaises(RuntimeError, newf_exception.result)
 
-        f_cancelled = self._new_future(loop=self.loop)
+        f_cancelled = concurrent.futures.Future()
         f_cancelled.cancel()
 
         newf_cancelled = self._new_future(loop=self.loop)
@@ -441,7 +441,7 @@ def test_copy_state(self):
         except BaseException as e:
             f_exc = e
 
-        f_conexc = self._new_future(loop=self.loop)
+        f_conexc = concurrent.futures.Future()
         f_conexc.set_exception(f_exc)
 
         newf_conexc = self._new_future(loop=self.loop)
@@ -454,6 +454,56 @@ def test_copy_state(self):
         newf_tb = ''.join(traceback.format_tb(newf_exc.__traceback__))
         self.assertEqual(newf_tb.count('raise concurrent.futures.InvalidStateError'), 1)
 
+    def test_copy_state_from_concurrent_futures(self):
+        """Test _copy_future_state from concurrent.futures.Future.
+
+        This tests the optimized path using _get_snapshot when available.
+        """
+        from asyncio.futures import _copy_future_state
+
+        # Test with a result
+        f_concurrent = concurrent.futures.Future()
+        f_concurrent.set_result(42)
+        f_asyncio = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent, f_asyncio)
+        self.assertTrue(f_asyncio.done())
+        self.assertEqual(f_asyncio.result(), 42)
+
+        # Test with an exception
+        f_concurrent_exc = concurrent.futures.Future()
+        f_concurrent_exc.set_exception(ValueError("test exception"))
+        f_asyncio_exc = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent_exc, f_asyncio_exc)
+        self.assertTrue(f_asyncio_exc.done())
+        with self.assertRaises(ValueError) as cm:
+            f_asyncio_exc.result()
+        self.assertEqual(str(cm.exception), "test exception")
+
+        # Test with cancelled state
+        f_concurrent_cancelled = concurrent.futures.Future()
+        f_concurrent_cancelled.cancel()
+        f_asyncio_cancelled = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent_cancelled, f_asyncio_cancelled)
+        self.assertTrue(f_asyncio_cancelled.cancelled())
+
+        # Test that destination already cancelled prevents copy
+        f_concurrent_result = concurrent.futures.Future()
+        f_concurrent_result.set_result(10)
+        f_asyncio_precancelled = self._new_future(loop=self.loop)
+        f_asyncio_precancelled.cancel()
+        _copy_future_state(f_concurrent_result, f_asyncio_precancelled)
+        self.assertTrue(f_asyncio_precancelled.cancelled())
+
+        # Test exception type conversion
+        f_concurrent_invalid = concurrent.futures.Future()
+        f_concurrent_invalid.set_exception(concurrent.futures.InvalidStateError("invalid"))
+        f_asyncio_invalid = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent_invalid, f_asyncio_invalid)
+        self.assertTrue(f_asyncio_invalid.done())
+        with self.assertRaises(asyncio.exceptions.InvalidStateError) as cm:
+            f_asyncio_invalid.result()
+        self.assertEqual(str(cm.exception), "invalid")
+
     def test_iter(self):
         fut = self._new_future(loop=self.loop)
 
@@ -700,6 +750,10 @@ def test_future_cancelled_exception_refcycles(self):
         self.assertIsNotNone(exc)
         self.assertListEqual(gc.get_referrers(exc), [])
 
+    def test_future_disallow_multiple_initialization(self):
+        f = self._new_future(loop=self.loop)
+        with self.assertRaises(RuntimeError, msg="is already initialized"):
+            f.__init__(loop=self.loop)
 
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
@@ -1041,33 +1095,6 @@ def __getattribute__(self, name):
             fut.add_done_callback(fut_callback_0)
             self.assertRaises(ReachableCode, fut.set_result, "boom")
 
-    def test_use_after_free_on_fut_context_0_with_evil__getattribute__(self):
-        # see: https://github.com/python/cpython/issues/125984
-
-        class EvilEventLoop(SimpleEvilEventLoop):
-            def call_soon(self, *args, **kwargs):
-                super().call_soon(*args, **kwargs)
-                raise ReachableCode
-
-            def __getattribute__(self, name):
-                if name == 'call_soon':
-                    # resets the future's event loop
-                    fut.__init__(loop=SimpleEvilEventLoop())
-                return object.__getattribute__(self, name)
-
-        evil_loop = EvilEventLoop()
-        with mock.patch.object(self, 'loop', evil_loop):
-            fut = self._new_future()
-            self.assertIs(fut.get_loop(), evil_loop)
-
-            fut_callback_0 = mock.Mock()
-            fut_context_0 = mock.Mock()
-            fut.add_done_callback(fut_callback_0, context=fut_context_0)
-            del fut_context_0
-            del fut_callback_0
-            self.assertRaises(ReachableCode, fut.set_result, "boom")
-
-
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
 class CFutureDoneCallbackTests(BaseFutureDoneCallbackTests,
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
index a3c5351fed0..a2f80edeb1d 100644
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -1218,6 +1218,38 @@ def gen():
         loop.advance_time(10)
         loop.run_until_complete(asyncio.wait([a, b]))
 
+    def test_wait_discards_awaited_by_for_pending(self):
+        # gh-152569: wait() must remove itself from the await-graph of every
+        # future once it returns, including futures that never resolved.
+        async def coro():
+            immortal = self.loop.create_future()
+            done = self.new_task(self.loop, asyncio.sleep(0))
+            await asyncio.wait({done, immortal},
+                               return_when=asyncio.FIRST_COMPLETED)
+            self.assertFalse(immortal._asyncio_awaited_by)
+            immortal.cancel()
+
+        self.loop.run_until_complete(self.new_task(self.loop, coro()))
+
+    def test_gather_discards_awaited_by_for_pending(self):
+        # gh-157213: a child outliving gather() must lose the awaited-by edge
+        async def fail():
+            raise ValueError
+
+        async def survivor():
+            await asyncio.Future()
+
+        async def coro():
+            t = self.new_task(self.loop, survivor())
+            with self.assertRaises(ValueError):
+                await asyncio.gather(t, fail())
+            self.assertFalse(t._asyncio_awaited_by)
+            t.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await t
+
+        self.loop.run_until_complete(self.new_task(self.loop, coro()))
+
     def test_wait_really_done(self):
         # there is possibility that some tasks in the pending list
         # became done but their callbacks haven't all been called yet
@@ -2115,6 +2147,8 @@ def test_shield_cancel_outer(self):
         test_utils.run_briefly(self.loop)
         self.assertTrue(outer.cancelled())
         self.assertEqual(0, 0 if outer._callbacks is None else len(outer._callbacks))
+        self.assertFalse(inner._asyncio_awaited_by)
+        self.assertTrue({f for f, _ctx in inner._callbacks or []} <= {asyncio.tasks._log_on_exception})
 
     def test_shield_cancel_outer_result(self):
         mock_handler = mock.Mock()
@@ -2140,6 +2174,21 @@ def test_shield_cancel_outer_exception(self):
         test_utils.run_briefly(self.loop)
         mock_handler.assert_called_once()
 
+    def test_shield_cancel_outer_in_task(self):
+        inner = self.new_future(self.loop)
+
+        async def coro():
+            outer = asyncio.shield(inner)
+            self.assertNotEqual(0, len(inner._callbacks))
+            outer.cancel()
+            await asyncio.sleep(0)
+            self.assertTrue(outer.cancelled())
+
+        task = self.new_task(self.loop, coro())
+        self.loop.run_until_complete(task)
+        self.assertFalse(inner._asyncio_awaited_by)
+        self.assertTrue({f for f, _ctx in inner._callbacks or []} <= {asyncio.tasks._log_on_exception})
+
     def test_shield_duplicate_log_once(self):
         mock_handler = mock.Mock()
         self.loop.set_exception_handler(mock_handler)
@@ -2581,6 +2630,68 @@ async def main():
         finally:
             loop.close()
 
+    def test_context_not_a_context(self):
+        # gh-157301
+        async def coro():
+            pass
+
+        loop = asyncio.new_event_loop()
+        c = coro()
+        try:
+            with self.assertRaises(TypeError):
+                self.new_task(loop, c, context='not a context')
+        finally:
+            c.close()
+            loop.close()
+
+    def test_context_not_a_context_leaves_loop_usable(self):
+        # gh-157301
+        async def coro():
+            pass
+
+        async def main():
+            c = coro()
+            try:
+                with self.assertRaises(TypeError):
+                    self.new_task(loop, c, context='not a context',
+                                  eager_start=True)
+            finally:
+                c.close()
+            await asyncio.sleep(0)
+
+        loop = asyncio.new_event_loop()
+        loop.call_later(support.SHORT_TIMEOUT, loop.stop)
+        try:
+            loop.run_until_complete(self.new_task(loop, main()))
+        finally:
+            loop.close()
+
+    def test_context_already_entered_leaves_loop_usable(self):
+        # gh-157301
+        async def coro():
+            pass
+
+        async def main():
+            ctx = contextvars.copy_context()
+
+            def inside():
+                c = coro()
+                try:
+                    with self.assertRaises(RuntimeError):
+                        self.new_task(loop, c, context=ctx, eager_start=True)
+                finally:
+                    c.close()
+
+            ctx.run(inside)
+            await asyncio.sleep(0)
+
+        loop = asyncio.new_event_loop()
+        loop.call_later(support.SHORT_TIMEOUT, loop.stop)
+        try:
+            loop.run_until_complete(self.new_task(loop, main()))
+        finally:
+            loop.close()
+
     def test_context_2(self):
         cvar = contextvars.ContextVar('cvar', default='nope')
 
@@ -2739,6 +2850,18 @@ async def main():
             self.assertEqual(name, "example")
             await t
 
+    def test_eager_start_true_no_loop(self):
+        # gh-154695: eager_start must use the resolved loop, not loop=None.
+        async def asyncfn():
+            return 42
+
+        async def main():
+            t = self.__class__.Task(asyncfn(), eager_start=True)
+            self.assertTrue(t.done())
+            self.assertEqual(await t, 42)
+
+        asyncio.run(main(), loop_factory=asyncio.EventLoop)
+
     def test_eager_start_false(self):
         name = None
 
@@ -2776,28 +2899,17 @@ def test_get_context(self):
         finally:
             loop.close()
 
-    def test_proper_refcounts(self):
-        # see: https://github.com/python/cpython/issues/126083
-        class Break:
-            def __str__(self):
-                raise RuntimeError("break")
-
-        obj = object()
-        initial_refcount = sys.getrefcount(obj)
-
-        coro = coroutine_function()
-        with contextlib.closing(asyncio.EventLoop()) as loop:
-            task = asyncio.Task.__new__(asyncio.Task)
-            for _ in range(5):
-                with self.assertRaisesRegex(RuntimeError, 'break'):
-                    task.__init__(coro, loop=loop, context=obj, name=Break())
-
-            coro.close()
-            task._log_destroy_pending = False
-            del task
+    def test_task_disallow_multiple_initialization(self):
+        async def foo():
+            pass
 
-            self.assertEqual(sys.getrefcount(obj), initial_refcount)
+        coro = foo()
+        self.addCleanup(coro.close)
+        task = self.new_task(self.loop, coro)
+        task._log_destroy_pending = False
 
+        with self.assertRaises(RuntimeError, msg="is already initialized"):
+            task.__init__(coro, loop=self.loop)
 
 def add_subclass_tests(cls):
     BaseTask = cls.Task
@@ -2921,19 +3033,6 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
-    @support.refcount_test
-    def test_refleaks_in_task___init__(self):
-        gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')
-        async def coro():
-            pass
-        task = self.new_task(self.loop, coro())
-        self.loop.run_until_complete(task)
-        refs_before = gettotalrefcount()
-        for i in range(100):
-            task.__init__(coro(), loop=self.loop)
-            self.loop.run_until_complete(task)
-        self.assertAlmostEqual(gettotalrefcount() - refs_before, 0, delta=10)
-
     def test_del__log_destroy_pending_segfault(self):
         async def coro():
             pass
@@ -2942,6 +3041,16 @@ async def coro():
         with self.assertRaises(AttributeError):
             del task._log_destroy_pending
 
+    def test_get_context_uninitialized_segfault(self):
+        # https://github.com/python/cpython/issues/154871
+
+        class UninitializedTask(self.Task):
+            def __init__(self, *args, **kwargs):
+                pass
+
+        task = UninitializedTask()
+        self.assertIsNone(task.get_context())
+
 
 @unittest.skipUnless(hasattr(futures, '_CFuture') and
                      hasattr(tasks, '_CTask'),
diff --git a/Lib/test/test_concurrent_futures/test_future.py b/Lib/test/test_concurrent_futures/test_future.py
index 4066ea1ee4b..06b11a3bacf 100644
--- a/Lib/test/test_concurrent_futures/test_future.py
+++ b/Lib/test/test_concurrent_futures/test_future.py
@@ -6,6 +6,7 @@
     PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future)
 
 from test import support
+from test.support import threading_helper
 
 from .util import (
     PENDING_FUTURE, RUNNING_FUTURE, CANCELLED_FUTURE,
@@ -282,6 +283,62 @@ def test_multiple_set_exception(self):
 
         self.assertEqual(f.exception(), e)
 
+    def test_get_snapshot(self):
+        """Test the _get_snapshot method for atomic state retrieval."""
+        # Test with a pending future
+        f = Future()
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertFalse(done)
+        self.assertFalse(cancelled)
+        self.assertIsNone(result)
+        self.assertIsNone(exception)
+
+        # Test with a finished future (successful result)
+        f = Future()
+        f.set_result(42)
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertTrue(done)
+        self.assertFalse(cancelled)
+        self.assertEqual(result, 42)
+        self.assertIsNone(exception)
+
+        # Test with a finished future (exception)
+        f = Future()
+        exc = ValueError("test error")
+        f.set_exception(exc)
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertTrue(done)
+        self.assertFalse(cancelled)
+        self.assertIsNone(result)
+        self.assertIs(exception, exc)
+
+        # Test with a cancelled future
+        f = Future()
+        f.cancel()
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertTrue(done)
+        self.assertTrue(cancelled)
+        self.assertIsNone(result)
+        self.assertIsNone(exception)
+
+        # Test concurrent access (basic thread safety check)
+        f = Future()
+        f.set_result(100)
+        results = []
+
+        def get_snapshot():
+            for _ in range(1000):
+                snapshot = f._get_snapshot()
+                results.append(snapshot)
+
+        threads = [threading.Thread(target=get_snapshot) for _ in range(4)]
+        with threading_helper.start_threads(threads):
+            pass
+        # All snapshots should be identical for a finished future
+        expected = (True, False, 100, None)
+        for result in results:
+            self.assertEqual(result, expected)
+
 
 def setUpModule():
     setup_module()
diff --git a/crates/stdlib/src/_asyncio.rs b/crates/stdlib/src/_asyncio.rs
index f56943c19d7..8b5c50d80ff 100644
--- a/crates/stdlib/src/_asyncio.rs
+++ b/crates/stdlib/src/_asyncio.rs
@@ -188,6 +188,13 @@ pub(crate) mod _asyncio {
             loop_: Option<PyObjectRef>,
             vm: &VirtualMachine,
         ) -> PyResult<()> {
+            if zelf.fut_loop.read().is_some() {
+                return Err(vm.new_runtime_error(format!(
+                    "{} object is already initialized",
+                    zelf.class().fully_qualified_name(vm)?
+                )));
+            }
+
             // Get the event loop
             let loop_obj = match loop_ {
                 Some(l) if !vm.is_none(&l) => l,
@@ -1183,6 +1190,13 @@ pub(crate) mod _asyncio {
     )]
     impl PyTask {
         fn py_init(zelf: &Py<Self>, args: TaskInitArgs, vm: &VirtualMachine) -> PyResult<()> {
+            if zelf.base.fut_loop.read().is_some() {
+                return Err(vm.new_runtime_error(format!(
+                    "{} object is already initialized",
+                    zelf.class().fully_qualified_name(vm)?
+                )));
+            }
+
             // Validate coroutine
             if !is_coroutine(args.coro.clone(), vm)? {
                 return Err(vm.new_type_error(format!(
@@ -1217,8 +1231,18 @@ pub(crate) mod _asyncio {
 
             // Get or create context
             let context = match args.context {
-                Some(c) => c,
-                None => get_copy_context(vm)?,
+                Some(c) if !vm.is_none(&c) => {
+                    if !c.class().is(PyContext::class(&vm.ctx)) {
+                        zelf.task_log_destroy_pending
+                            .store(false, Ordering::Relaxed);
+                        return Err(vm.new_type_error(format!(
+                            "a contextvars.Context was expected, got {}",
+                            c.class().fully_qualified_name(vm)?
+                        )));
+                    }
+                    c
+                }
+                _ => get_copy_context(vm)?,
             };
             *zelf.task_context.write() = Some(context);
 
@@ -2043,7 +2067,25 @@ pub(crate) mod _asyncio {
                     match ctx.downcast::<PyContext>() {
                         Ok(ctx) => {
                             // Only exit a context that was actually entered.
-                            PyContext::enter(&ctx, vm)?;
+                            if let Err(error) = PyContext::enter(&ctx, vm) {
+                                zelf.task_log_destroy_pending
+                                    .store(false, Ordering::Relaxed);
+                                let _ = _swap_current_task(
+                                    LoopTaskArgs {
+                                        loop_: loop_obj,
+                                        task: prev_task,
+                                    },
+                                    vm,
+                                );
+                                let _ = _unregister_eager_task(
+                                    TaskArg {
+                                        task: task_obj.clone(),
+                                    },
+                                    vm,
+                                );
+                                let _ = _unregister_task(TaskArg { task: task_obj }, vm);
+                                return Err(error);
+                            }
                             let result = vm.call_method(&c, "send", (vm.ctx.none(),));
                             PyContext::exit(&ctx, vm)?;
                             result
diff --git a/extra_tests/snippets/stdlib_asyncio_reinitialization_315.py b/extra_tests/snippets/stdlib_asyncio_reinitialization_315.py
new file mode 100644
index 00000000000..0e6e1ea9041
--- /dev/null
+++ b/extra_tests/snippets/stdlib_asyncio_reinitialization_315.py
@@ -0,0 +1,75 @@
+"""A rejected native Task reinitialization must preserve the pending task."""
+
+import asyncio
+from asyncio import tasks
+
+
+class Loop:
+    def __init__(self):
+        self.ready = []
+
+    def get_debug(self):
+        return False
+
+    def is_running(self):
+        return False
+
+    def call_soon(self, callback, *args, context=None):
+        self.ready.append((callback, args, context))
+
+    def call_exception_handler(self, context):
+        raise AssertionError(context)
+
+    def step(self):
+        callback, args, context = self.ready.pop(0)
+        previous = asyncio.events._get_running_loop()
+        asyncio.events._set_running_loop(self)
+        try:
+            if context is None:
+                callback(*args)
+            else:
+                context.run(callback, *args)
+        finally:
+            asyncio.events._set_running_loop(previous)
+
+
+def rejected(callback):
+    try:
+        callback()
+    except RuntimeError:
+        pass
+    else:
+        raise AssertionError("reinitialization was accepted")
+
+
+def test_task_reinitialization_preserves_pending_task():
+    loop = Loop()
+    token = object()
+
+    async def original():
+        return token
+
+    async def replacement():
+        raise AssertionError("replacement coroutine must never run")
+
+    coro, other = original(), replacement()
+    t = tasks._CTask(coro, loop=loop, name="original")
+    t._log_destroy_pending = False
+    before = (t.done(), t.cancelled(), t.cancelling(), len(loop.ready))
+    try:
+        rejected(lambda: t.__init__(other, loop=Loop(), name="replacement"))
+        # Rejection must precede coroutine validation on an initialized task.
+        rejected(lambda: t.__init__(object(), loop=Loop()))
+        assert t.get_loop() is loop
+        assert t.get_coro() is coro
+        assert t.get_name() == "original"
+        assert (t.done(), t.cancelled(), t.cancelling(), len(loop.ready)) == before
+        assert len(loop.ready) == 1
+        loop.step()
+        assert t.result() is token
+    finally:
+        other.close()
+        coro.close()
+
+
+test_task_reinitialization_preserves_pending_task()

From 55f925885f55c8453b46473ae26098913b486ce6 Mon Sep 17 00:00:00 2001
From: youknowdot <dot@youknowone.org>
Date: Thu, 15 Oct 2026 15:53:41 +0900
Subject: [PATCH 2/3] Remove asyncio snippet duplicating canonical
 reinitialization coverage

Keep the exact-final Lib tests as the reinitialization coverage instead of adding a new snippet for the same contract. Native and library implementation bytes remain unchanged.

Assisted-by: OpenAI Codex:model version unavailable
---
 .../stdlib_asyncio_reinitialization_315.py    | 75 -------------------
 1 file changed, 75 deletions(-)
 delete mode 100644 extra_tests/snippets/stdlib_asyncio_reinitialization_315.py

diff --git a/extra_tests/snippets/stdlib_asyncio_reinitialization_315.py b/extra_tests/snippets/stdlib_asyncio_reinitialization_315.py
deleted file mode 100644
index 0e6e1ea9041..00000000000
--- a/extra_tests/snippets/stdlib_asyncio_reinitialization_315.py
+++ /dev/null
@@ -1,75 +0,0 @@
-"""A rejected native Task reinitialization must preserve the pending task."""
-
-import asyncio
-from asyncio import tasks
-
-
-class Loop:
-    def __init__(self):
-        self.ready = []
-
-    def get_debug(self):
-        return False
-
-    def is_running(self):
-        return False
-
-    def call_soon(self, callback, *args, context=None):
-        self.ready.append((callback, args, context))
-
-    def call_exception_handler(self, context):
-        raise AssertionError(context)
-
-    def step(self):
-        callback, args, context = self.ready.pop(0)
-        previous = asyncio.events._get_running_loop()
-        asyncio.events._set_running_loop(self)
-        try:
-            if context is None:
-                callback(*args)
-            else:
-                context.run(callback, *args)
-        finally:
-            asyncio.events._set_running_loop(previous)
-
-
-def rejected(callback):
-    try:
-        callback()
-    except RuntimeError:
-        pass
-    else:
-        raise AssertionError("reinitialization was accepted")
-
-
-def test_task_reinitialization_preserves_pending_task():
-    loop = Loop()
-    token = object()
-
-    async def original():
-        return token
-
-    async def replacement():
-        raise AssertionError("replacement coroutine must never run")
-
-    coro, other = original(), replacement()
-    t = tasks._CTask(coro, loop=loop, name="original")
-    t._log_destroy_pending = False
-    before = (t.done(), t.cancelled(), t.cancelling(), len(loop.ready))
-    try:
-        rejected(lambda: t.__init__(other, loop=Loop(), name="replacement"))
-        # Rejection must precede coroutine validation on an initialized task.
-        rejected(lambda: t.__init__(object(), loop=Loop()))
-        assert t.get_loop() is loop
-        assert t.get_coro() is coro
-        assert t.get_name() == "original"
-        assert (t.done(), t.cancelled(), t.cancelling(), len(loop.ready)) == before
-        assert len(loop.ready) == 1
-        loop.step()
-        assert t.result() is token
-    finally:
-        other.close()
-        coro.close()
-
-
-test_task_reinitialization_preserves_pending_task()

From 33715c9853b765a333278fa4418c1396b4f2943a Mon Sep 17 00:00:00 2001
From: youknowdot <dot@youknowone.org>
Date: Thu, 15 Oct 2026 16:32:57 +0900
Subject: [PATCH 3/3] Validate both objects in asyncio awaited-by helpers

Match CPython by ignoring non-native Future/Task waiters in add and discard. Keep the existing native reentrant hashing fixture reachable using Future subclasses and use native instances for its membership checks.

Assisted-by: OpenAI Codex:model version unavailable
---
 crates/stdlib/src/_asyncio.rs          | 12 ++++++++++--
 extra_tests/snippets/stdlib_asyncio.py | 12 ++++++------
 2 files changed, 16 insertions(+), 8 deletions(-)

diff --git a/crates/stdlib/src/_asyncio.rs b/crates/stdlib/src/_asyncio.rs
index 8b5c50d80ff..8da4ab38da5 100644
--- a/crates/stdlib/src/_asyncio.rs
+++ b/crates/stdlib/src/_asyncio.rs
@@ -2881,8 +2881,12 @@ pub(crate) mod _asyncio {
         waiter: PyObjectRef,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
-        // Only operate on native Future/Task objects (including subclasses).
+        // Both arguments must be native Future/Task objects (including subclasses).
         // Non-native objects are silently ignored.
+        if waiter.downcast_ref::<PyTask>().is_none() && waiter.downcast_ref::<PyFuture>().is_none()
+        {
+            return Ok(());
+        }
         if let Some(task) = fut.downcast_ref::<PyTask>() {
             return task.base.awaited_by_add(waiter, vm);
         }
@@ -2898,8 +2902,12 @@ pub(crate) mod _asyncio {
         waiter: PyObjectRef,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
-        // Only operate on native Future/Task objects (including subclasses).
+        // Both arguments must be native Future/Task objects (including subclasses).
         // Non-native objects are silently ignored.
+        if waiter.downcast_ref::<PyTask>().is_none() && waiter.downcast_ref::<PyFuture>().is_none()
+        {
+            return Ok(());
+        }
         if let Some(task) = fut.downcast_ref::<PyTask>() {
             return task.base.awaited_by_discard(&waiter, vm);
         }
diff --git a/extra_tests/snippets/stdlib_asyncio.py b/extra_tests/snippets/stdlib_asyncio.py
index a6a55509036..74de3604073 100644
--- a/extra_tests/snippets/stdlib_asyncio.py
+++ b/extra_tests/snippets/stdlib_asyncio.py
@@ -76,9 +76,9 @@ def __new__(cls, *args):
 # to the same future; the field must not be locked while that runs.
 
 
-class Reentrant:
+class Reentrant(_asyncio.Future):
     def __hash__(self):
-        _asyncio.future_add_to_awaited_by(awaited, Reentrant())
+        _asyncio.future_add_to_awaited_by(awaited, Reentrant(loop=object()))
         return 1
 
     def __eq__(self, other):
@@ -86,15 +86,15 @@ def __eq__(self, other):
 
 
 awaited = _asyncio.Future(loop=object())
-_asyncio.future_add_to_awaited_by(awaited, Reentrant())
+_asyncio.future_add_to_awaited_by(awaited, Reentrant(loop=object()))
 with assert_raises(RecursionError):
     # converting the single waiter into a set hashes both of them
-    _asyncio.future_add_to_awaited_by(awaited, Reentrant())
+    _asyncio.future_add_to_awaited_by(awaited, Reentrant(loop=object()))
 
 plain = _asyncio.Future(loop=object())
-waiter = object()
+waiter = _asyncio.Future(loop=object())
 _asyncio.future_add_to_awaited_by(plain, waiter)
-_asyncio.future_add_to_awaited_by(plain, object())
+_asyncio.future_add_to_awaited_by(plain, _asyncio.Future(loop=object()))
 assert waiter in plain._asyncio_awaited_by
 _asyncio.future_discard_from_awaited_by(plain, waiter)
 assert waiter not in plain._asyncio_awaited_by