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..8da4ab38da5 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
@@ -2839,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);
         }
@@ -2856,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