gh-158406: Fix run_until_complete making the next run stop early

Lib/asyncio/base_events.py

@@ -707,8 +707,19 @@ def run_until_complete(self, future):
             # is no need to log the "destroy pending task" message
             future._log_destroy_pending = False
 
-        future.add_done_callback(_run_until_complete_cb)
+        # gh-158406: A queued callback can outlive this call. Only let
+        # it stop the loop while this call is running.
+        running = True
+
+        def done_cb(fut):
+            if running:
+                _run_until_complete_cb(fut)
+            elif not fut.cancelled():
+                # Still mark the exception as retrieved.
+                fut.exception()
+
         try:
+            future.add_done_callback(done_cb)
             self.run_forever()
         except:
             if new_task and future.done() and not future.cancelled():
@@ -718,7 +729,8 @@ def run_until_complete(self, future):
                 future.exception()
             raise
         finally:
-            future.remove_done_callback(_run_until_complete_cb)
+            running = False
+            future.remove_done_callback(done_cb)
         if not future.done():
             raise RuntimeError('Event loop stopped before Future completed.')
 

Lib/test/test_asyncio/test_base_events.py

@@ -918,6 +918,56 @@ def func():
         self.loop.run_forever()
         self.assertTrue(func.called)
 
+    def test_run_until_complete_interrupted_after_done(self):
+        # gh-158406: an interrupt after the future is done must not stop
+        # the next run_until_complete() call early.
+        self.loop._process_events = mock.Mock()
+        for exc in (KeyboardInterrupt, SystemExit):
+            with self.subTest(exc=exc):
+                def interrupt():
+                    raise exc
+
+                async def main():
+                    self.loop.call_soon(interrupt)
+
+                with self.assertRaises(exc):
+                    self.loop.run_until_complete(main())
+                result = self.loop.run_until_complete(asyncio.sleep(0, "ret"))
+                self.assertEqual(result, "ret")
+
+    def test_run_until_complete_stopped_after_done(self):
+        # gh-158406: stop() in the iteration in which the future is done
+        # must not stop the next run_until_complete() call early.
+        async def main():
+            self.loop.stop()
+            return 1
+
+        self.loop._process_events = mock.Mock()
+        self.assertEqual(self.loop.run_until_complete(main()), 1)
+        result = self.loop.run_until_complete(asyncio.sleep(0, "ret"))
+        self.assertEqual(result, "ret")
+
+    def test_run_forever_after_interrupted_run_until_complete(self):
+        # gh-158406: an interrupt after the future is done must not stop
+        # a later run_forever() call early.
+        def interrupt():
+            raise KeyboardInterrupt
+
+        async def main():
+            self.loop.call_soon(interrupt)
+
+        self.loop._process_events = mock.Mock()
+        with self.assertRaises(KeyboardInterrupt):
+            self.loop.run_until_complete(main())
+
+        def func():
+            self.loop.stop()
+            func.called = True
+        func.called = False
+        self.loop.call_soon(self.loop.call_soon, func)
+        self.loop.run_forever()
+        self.assertTrue(func.called)
+
     def test_single_selecter_event_callback_after_stopping(self):
         # Python issue #25593: A stopped event loop may cause event callbacks
         # to run more than once.

Lib/test/test_asyncio/test_runners.py

@@ -461,6 +461,20 @@ async def coro():
             with self.assertRaises(asyncio.CancelledError):
                 runner.run(coro())
 
+    def test_interrupt_after_main_task_done(self):
+        assert threading.current_thread() is threading.main_thread()
+
+        async def coro():
+            nonlocal task
+            task = asyncio.create_task(asyncio.Event().wait())
+            asyncio.get_running_loop().call_soon(interrupt_self)
+
+        task = None
+        with asyncio.Runner(loop_factory=self.new_loop) as runner:
+            with self.assertRaises(KeyboardInterrupt):
+                runner.run(coro())
+        self.assertTrue(task.cancelled())
+
     def test_signal_install_not_supported_ok(self):
         # signal.signal() can throw if the "main thread" doesn't have signals enabled
         assert threading.current_thread() is threading.main_thread()

Misc/NEWS.d/next/Library/2026-09-29-20-42-23.gh-issue-158406.QdOXgn.rst

@@ -0,0 +1,4 @@
+Fix :meth:`asyncio.loop.run_until_complete` raising :exc:`RuntimeError` when
+the previous call was interrupted or stopped just as its future completed.
+This could make :func:`asyncio.run` raise :exc:`RuntimeError` instead of
+:exc:`KeyboardInterrupt` on Ctrl-C and skip part of its cleanup.