gh-159041: Fix asyncio task tracking during thread state cleanup

Move lingering asyncio tasks to the interpreter only after all
finalizer-capable thread-state cleanup has completed. Also transfer tasks
created by finalizers during interpreter shutdown.

Add regression tests for both thread-state and interpreter cleanup paths.

Author: ngoldbaum

Lib/test/test_asyncio/test_tasks.py

@@ -1,5 +1,6 @@
 """Tests for tasks.py."""
 
+import _thread
 import collections
 import contextlib
 import contextvars
@@ -20,6 +21,7 @@
 from asyncio import tasks
 from test.test_asyncio import utils as test_utils
 from test import support
+from test.support import threading_helper
 from test.support.script_helper import assert_python_ok
 
 
@@ -3095,6 +3097,68 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
+    @threading_helper.requires_working_threading()
+    def test_task_created_during_thread_state_clear(self):
+        loop = self.loop
+        task = None
+        var = contextvars.ContextVar('var')
+        task_context = contextvars.Context()
+
+        async def noop():
+            pass
+
+        class CreatesTaskOnClear:
+            def __del__(self):
+                nonlocal task
+                task = loop.create_task(noop(), context=task_context)
+
+        def create_finalizer():
+            var.set(CreatesTaskOnClear())
+
+        handle = _thread.start_joinable_thread(create_finalizer)
+        handle.join(support.SHORT_TIMEOUT)
+        self.assertTrue(handle.is_done())
+        self.assertIsNotNone(task)
+
+        try:
+            self.assertEqual(self.all_tasks(loop), {task})
+        finally:
+            loop.run_until_complete(task)
+
+    def test_task_created_during_interpreter_clear(self):
+        code = """if 1:
+            import asyncio
+            import contextvars
+            import warnings
+
+            async def noop():
+                pass
+
+            class Loop:
+                def get_debug(self):
+                    return False
+
+                def call_soon(self, callback, *args, context=None):
+                    self.callback = callback
+
+            class CreatesTaskOnClear:
+                def __init__(self):
+                    self.task_type = asyncio.Task
+                    self.coro = noop()
+                    self.loop = Loop()
+                    self.context = contextvars.Context()
+
+                def __del__(self):
+                    self.task_type(
+                        self.coro, loop=self.loop, context=self.context)
+
+            # Leave the only reference to the filters list in the warnings
+            # state, which is cleared after the final garbage collection.
+            warnings.filters.append(CreatesTaskOnClear())
+            del warnings.filters
+        """
+        assert_python_ok('-c', code)
+
     def test_del__log_destroy_pending_segfault(self):
         async def coro():
             pass

Misc/NEWS.d/next/Core_and_Builtins/2026-10-09-16-04-40.gh-issue-159041.PUP8bU.rst

@@ -0,0 +1,2 @@
+Fix a use-after-free in :mod:`asyncio` task bookkeeping when a finalizer
+creates a task while its thread state is being cleared.

Python/pystate.c

@@ -812,6 +812,16 @@ common_constants_clear(PyInterpreterState *interp)
 }
 
 
+static void
+move_asyncio_tasks_to_interpreter(PyThreadState *tstate)
+{
+    PyMutex_Lock(&tstate->interp->asyncio_tasks_lock);
+    llist_concat(&tstate->interp->asyncio_tasks_head,
+                 &((_PyThreadStateImpl *)tstate)->asyncio_tasks_head);
+    PyMutex_Unlock(&tstate->interp->asyncio_tasks_lock);
+}
+
+
 static void
 interpreter_clear(PyInterpreterState *interp, PyThreadState *tstate)
 {
@@ -953,6 +963,10 @@ interpreter_clear(PyInterpreterState *interp, PyThreadState *tstate)
     if (tstate->interp == interp) {
         /* We are now safe to fix tstate->_status.cleared. */
         // XXX Do this (much) earlier?
+        // Finalizers above may have registered new asyncio tasks.
+        move_asyncio_tasks_to_interpreter(tstate);
+        assert(llist_empty(
+            &((_PyThreadStateImpl *)tstate)->asyncio_tasks_head));
         tstate->_status.cleared = 1;
     }
 
@@ -1837,13 +1851,6 @@ PyThreadState_Clear(PyThreadState *tstate)
     Py_CLEAR(((_PyThreadStateImpl *)tstate)->asyncio_running_task);
 
 
-    PyMutex_Lock(&tstate->interp->asyncio_tasks_lock);
-    // merge any lingering tasks from thread state to interpreter's
-    // tasks list
-    llist_concat(&tstate->interp->asyncio_tasks_head,
-                 &((_PyThreadStateImpl *)tstate)->asyncio_tasks_head);
-    PyMutex_Unlock(&tstate->interp->asyncio_tasks_lock);
-
     Py_CLEAR(tstate->dict);
     Py_CLEAR(tstate->async_exc);
 
@@ -1910,6 +1917,11 @@ PyThreadState_Clear(PyThreadState *tstate)
     _PyJit_TracerFree((_PyThreadStateImpl *)tstate);
 #endif
 
+    // Merge any lingering tasks from the thread state to the interpreter's
+    // tasks list.  This must happen after all cleanup which can run finalizers,
+    // since those finalizers may create and register new tasks.
+    move_asyncio_tasks_to_interpreter(tstate);
+
     tstate->_status.cleared = 1;
 
     // XXX Call _PyThreadStateSwap(runtime, NULL) here if "current".