diff --git a/Doc/includes/asyncio_guest_tkinter.py b/Doc/includes/asyncio_guest_tkinter.py
new file mode 100644
index 00000000000000..356fbe46b5a0a6
--- /dev/null
+++ b/Doc/includes/asyncio_guest_tkinter.py
@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""Minimal demo: asyncio running as a guest inside Tkinter's mainloop.
+
+A progress bar counts from 0 to MAX_COUNT using ``asyncio.sleep()``.
+The Tk GUI stays fully responsive throughout.  Closing the window or
+pressing the Cancel button cancels the async task cleanly.
+
+Usage::
+
+    python asyncio_guest_tkinter.py
+"""
+
+import asyncio
+import collections
+import tkinter as tk
+import tkinter.ttk as ttk
+import traceback
+
+
+# -- Host adapter for Tkinter ------------------------------------------
+
+class TkHost:
+    """Bridge between asyncio guest mode and the Tk event loop."""
+
+    def __init__(self, root):
+        self.root = root
+        self._tk_func_name = root.register(self._dispatch)
+        self._q = collections.deque()
+
+    def _dispatch(self):
+        self._q.popleft()()
+
+    def run_sync_soon_threadsafe(self, fn):
+        """Schedule *fn* on the Tk thread.
+
+        ``Tkapp_ThreadSend`` (the C layer behind ``root.call`` from a
+        non-Tcl thread) posts the command to the Tcl event queue, making
+        this safe to call from any thread.
+        """
+        self._q.append(fn)
+        self.root.call('after', 'idle', self._tk_func_name)
+
+    def done_callback(self, task):
+        """Called when the async task finishes."""
+        if task.cancelled():
+            print("Task was cancelled.")
+        elif task.exception() is not None:
+            exc = task.exception()
+            traceback.print_exception(type(exc), exc, exc.__traceback__)
+        else:
+            print(f"Task returned: {task.result()}")
+        self.root.destroy()
+
+
+# -- Async workload ----------------------------------------------------
+
+MAX_COUNT = 20
+PERIOD = 0.5  # seconds between increments
+
+
+async def count(progress, root):
+    """Increment a progress bar, updating the Tk GUI each step."""
+    root.wm_title(f"Counting every {PERIOD}s ...")
+    progress.configure(maximum=MAX_COUNT)
+
+    task = asyncio.current_task()
+    loop = asyncio.get_running_loop()
+
+    # Wire the Cancel button and window close to task.cancel().
+    # Use call_soon_threadsafe so the I/O thread's selector is woken.
+    def request_cancel():
+        loop.call_soon_threadsafe(task.cancel)
+
+    cancel_btn = root.nametowidget('cancel')
+    cancel_btn.configure(command=request_cancel)
+    root.protocol("WM_DELETE_WINDOW", request_cancel)
+
+    for i in range(1, MAX_COUNT + 1):
+        await asyncio.sleep(PERIOD)
+        progress.step(1)
+        root.wm_title(f"Count: {i}/{MAX_COUNT}")
+
+    return i
+
+
+# -- Main ---------------------------------------------------------------
+
+def main():
+    root = tk.Tk()
+    root.wm_title("asyncio guest + Tkinter")
+
+    progress = ttk.Progressbar(root, length='6i')
+    progress.pack(fill=tk.BOTH, expand=True, padx=8, pady=(8, 4))
+
+    cancel_btn = tk.Button(root, text='Cancel', name='cancel')
+    cancel_btn.pack(pady=(0, 8))
+
+    host = TkHost(root)
+
+    asyncio.start_guest_run(
+        count, progress, root,
+        run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
+        done_callback=host.done_callback,
+    )
+
+    root.mainloop()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst
index 41abb2d7d0a53e..ead73390b2ac28 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/Doc/library/asyncio-eventloop.rst
@@ -207,6 +207,44 @@ Running and stopping the loop
    .. versionchanged:: 3.12
       Added the *timeout* parameter.
 
+Decomposing event loop iteration
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The following methods decompose a single iteration of the event loop
+into independently callable steps.  They are used internally by
+:func:`asyncio.start_guest_run`; see :ref:`asyncio-guest` for full
+documentation.
+
+.. method:: loop.poll_events()
+
+   Poll for I/O events without processing them.
+
+   Cleans up cancelled scheduled handles, computes an appropriate
+   timeout from the scheduled callbacks, and calls the underlying
+   selector.  Returns the raw event list.
+
+   .. versionadded:: 3.16
+
+.. method:: loop.process_events(event_list)
+
+   Process I/O events returned by :meth:`poll_events`.
+
+   Delegates to the selector-specific event processing that turns raw
+   selector events into ready callbacks.
+
+   .. versionadded:: 3.16
+
+.. method:: loop.process_ready()
+
+   Process expired timers and execute ready callbacks.
+
+   Moves scheduled callbacks whose deadline has passed into the ready
+   queue, then runs all callbacks that were ready at call time.
+   Callbacks enqueued *by* running callbacks are left for the next
+   iteration.
+
+   .. versionadded:: 3.16
+
 Scheduling callbacks
 ^^^^^^^^^^^^^^^^^^^^
 
diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst
new file mode 100644
index 00000000000000..cfffb85149ca06
--- /dev/null
+++ b/Doc/library/asyncio-guest.rst
@@ -0,0 +1,150 @@
+.. currentmodule:: asyncio
+
+.. _asyncio-guest:
+
+==========
+Guest Mode
+==========
+
+**Source code:** :source:`Lib/asyncio/guest.py`
+
+----
+
+Running asyncio as a Guest in Another Event Loop
+=================================================
+
+*Guest mode* allows asyncio to run cooperatively inside a *host* event loop
+such as a GUI toolkit's main loop (Tkinter, Qt, GTK, etc.).  Instead of
+replacing the host loop, asyncio piggybacks on it:
+
+* The **host thread** keeps running its own main loop as usual.
+* A **background I/O thread** blocks on the selector (I/O polling).
+  When I/O events arrive it hands them back to the host thread via a
+  thread-safe callback.  The thread is not a daemon thread; it is joined
+  when the guest run finishes.
+* The host thread then runs
+  :meth:`loop.process_events() <asyncio.loop.process_events>` and
+  :meth:`loop.process_ready() <asyncio.loop.process_ready>` to advance
+  the asyncio event loop by one step, then signals the I/O thread to
+  poll again.
+
+Exactly one of the two threads touches the event loop at any moment, so
+neither the host loop nor the asyncio loop starves the other.
+
+Typical use cases:
+
+* Incrementally migrating a Tkinter/Qt/GTK application to ``async/await``
+  without replacing the native event loop.
+* Embedding asyncio I/O (HTTP clients, websockets, …) inside a GUI app.
+* Running asyncio alongside a framework that owns the main thread.
+
+.. rubric:: Example
+
+See :source:`Doc/includes/asyncio_guest_tkinter.py` for a complete Tkinter
+example that embeds asyncio inside ``tkinter.mainloop()`` using
+:func:`start_guest_run`.
+
+.. seealso::
+
+   The `asyncio-guest <https://github.com/congzhangzh/asyncio-guest>`__
+   project — the proof of concept this feature is based on — has runnable
+   examples for many more hosts: Tkinter, Qt (PyQt5/PySide6), GTK,
+   pygame, Win32 and Tornado.
+
+   `Trio's guest mode
+   <https://trio.readthedocs.io/en/stable/reference-lowlevel.html#using-guest-mode-to-run-trio-on-top-of-other-event-loops>`__,
+   which pioneered this approach.
+
+.. rubric:: API
+
+.. function:: start_guest_run(async_fn, *args, run_sync_soon_threadsafe, done_callback)
+
+   Run *async_fn* as a guest inside another event loop.
+
+   Must be called from the host event loop's thread.  The host loop
+   (e.g. ``tkinter.mainloop()``) remains in control of that thread;
+   asyncio I/O polling runs in a background non-daemon thread that is
+   joined when the run finishes.
+
+   :param async_fn: The async function to run as the top-level coroutine.
+   :param args: Positional arguments forwarded to *async_fn*.
+   :param run_sync_soon_threadsafe: A callable that schedules a zero-argument
+       callable on the host event loop's thread.  It must be thread-safe,
+       must not block, and must not raise; it need not preserve ordering.
+       For Tkinter use a ``root.call('after', 'idle', ...)`` wrapper; for
+       Qt use a ``QMetaObject.invokeMethod`` wrapper; etc.
+   :param done_callback: Called on the host thread after the run has fully
+       finished and the loop is closed (see :ref:`asyncio-guest-lifecycle`).
+       Receives the :class:`Task` as its sole argument.  Inspect the
+       outcome with :meth:`Task.result`, :meth:`Task.exception`, or
+       :meth:`Task.cancelled`.
+   :returns: The :class:`Task` wrapping *async_fn*.
+
+   To cancel the task from the host, use::
+
+       loop.call_soon_threadsafe(task.cancel)
+
+   This wakes the I/O thread from its selector wait so cancellation is
+   processed promptly.
+
+   .. versionadded:: 3.16
+
+.. _asyncio-guest-lifecycle:
+
+Lifecycle and Cleanup
+=====================
+
+For the whole guest run the guest loop is the host thread's running
+loop: :func:`get_running_loop` works inside guest tasks,
+:meth:`loop.is_running() <asyncio.loop.is_running>` returns ``True``, and
+starting another event loop on that thread — including a nested
+:func:`asyncio.run` or :meth:`loop.run_until_complete` — raises
+:exc:`RuntimeError`.  Consequently a thread that is already running an
+asyncio event loop cannot start a guest run.
+
+When the main task finishes, cleanup equivalent to :func:`asyncio.run`
+takes place on the host thread: remaining tasks are cancelled,
+asynchronous generators and the default executor are shut down, the I/O
+thread is joined, and the loop is closed.  Only then is *done_callback*
+invoked.
+
+If the interpreter exits while a guest run is unfinished, the run is
+abandoned: the I/O thread is woken and joined so that interpreter
+shutdown does not hang, pending tasks are not cancelled, and
+*done_callback* is not called.
+
+Signal Handling
+===============
+
+In guest mode the *host* owns signal handling:
+
+* The guest loop never touches :func:`signal.set_wakeup_fd`, neither to
+  install a file descriptor nor to reset it on close, so the host's
+  signal wake-up pipeline stays intact.
+* :meth:`loop.add_signal_handler` and :meth:`loop.remove_signal_handler`
+  raise :exc:`RuntimeError`.
+* To let asyncio code react to a signal, catch it in the host (with
+  :func:`signal.signal` or the host framework's facilities) and forward
+  it into the loop with :meth:`loop.call_soon_threadsafe`.
+
+Host Requirements
+=================
+
+* *run_sync_soon_threadsafe* must be thread-safe, non-blocking, and must
+  not raise.  It may run callbacks in any order.
+* Host code running *outside* guest callbacks (for example a GUI button
+  handler) must interact with the loop exclusively through
+  :meth:`loop.call_soon_threadsafe`, even though it runs on the loop's
+  own thread: the I/O thread may be inside the selector, and only
+  ``call_soon_threadsafe`` wakes it safely.
+* :meth:`loop.stop` is not supported in guest mode.
+
+.. rubric:: Low-level Event Loop Methods
+
+:func:`start_guest_run` drives the loop through three low-level methods
+-- :meth:`loop.poll_events() <asyncio.loop.poll_events>`,
+:meth:`loop.process_events() <asyncio.loop.process_events>`, and
+:meth:`loop.process_ready() <asyncio.loop.process_ready>` -- which
+decompose a single iteration of the event loop into independently
+callable steps.  See :ref:`asyncio-event-loop` for their reference
+documentation.
diff --git a/Doc/library/asyncio.rst b/Doc/library/asyncio.rst
index 956b00f0873a0d..13a7f911826eaa 100644
--- a/Doc/library/asyncio.rst
+++ b/Doc/library/asyncio.rst
@@ -132,6 +132,7 @@ for full functionality and the latest features.
    asyncio-protocol.rst
    asyncio-platforms.rst
    asyncio-extending.rst
+   asyncio-guest.rst
 
 .. toctree::
    :caption: Guides and Tutorials
diff --git a/Doc/whatsnew/3.16.rst b/Doc/whatsnew/3.16.rst
index 063755e1eadcb5..1dfc5f5176f078 100644
--- a/Doc/whatsnew/3.16.rst
+++ b/Doc/whatsnew/3.16.rst
@@ -109,6 +109,11 @@ asyncio
   socket file created for *path*.
   (Contributed by Sam Bull in :gh:`94984`.)
 
+* Add :func:`asyncio.start_guest_run` to run asyncio cooperatively inside
+  a host event loop, such as a GUI toolkit's main loop, that owns the
+  thread.  See :ref:`asyncio-guest`.
+  (Contributed by Cong Zhang in :gh:`145342`.)
+
 
 codecs
 ------
diff --git a/Lib/asyncio/__init__.py b/Lib/asyncio/__init__.py
index 2432e2dad74c23..b21fe969153bbe 100644
--- a/Lib/asyncio/__init__.py
+++ b/Lib/asyncio/__init__.py
@@ -11,6 +11,7 @@
 from .exceptions import *
 from .futures import *
 from .graph import *
+from .guest import *
 from .locks import *
 from .protocols import *
 from .runners import *
@@ -29,6 +30,7 @@
            exceptions.__all__ +
            futures.__all__ +
            graph.__all__ +
+           guest.__all__ +
            locks.__all__ +
            protocols.__all__ +
            runners.__all__ +
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index f26fba175b63cd..92a47fca698514 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -432,6 +432,10 @@ def __init__(self):
         # Identifier of the thread running the event loop, or None if the
         # event loop is not running
         self._thread_id = None
+        # True while the loop is driven as a guest inside a host event
+        # loop (see asyncio.start_guest_run); signal handling is then
+        # delegated to the host.
+        self._guest_mode = False
         self._clock_resolution = time.get_clock_info('monotonic').resolution
         self._exception_handler = None
         self.set_debug(coroutines._is_debug_mode())
@@ -1989,14 +1993,20 @@ def _timer_handle_cancelled(self, handle):
         if handle._scheduled:
             self._timer_cancelled_count += 1
 
-    def _run_once(self):
-        """Run one full iteration of the event loop.
+    def poll_events(self):
+        """Poll for I/O events without processing them.
 
-        This calls all currently ready callbacks, polls for I/O,
-        schedules the resulting callbacks, and finally schedules
-        'call_later' callbacks.
-        """
+        Cleans up cancelled scheduled handles, computes an appropriate
+        timeout from the scheduled callbacks, and calls
+        ``self._selector.select(timeout)``.  Returns the raw event list.
 
+        This method, together with :meth:`process_events` and
+        :meth:`process_ready`, decomposes :meth:`_run_once` into
+        independently callable steps so that an external event loop can
+        drive asyncio (see :func:`asyncio.start_guest_run`).
+
+        .. versionadded:: 3.16
+        """
         sched_count = len(self._scheduled)
         if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
             self._timer_cancelled_count / sched_count >
@@ -2031,11 +2041,29 @@ def _run_once(self):
             elif timeout < 0:
                 timeout = 0
 
-        event_list = self._selector.select(timeout)
+        return self._selector.select(timeout)
+
+    def process_events(self, event_list):
+        """Process I/O events returned by :meth:`poll_events`.
+
+        Delegates to the selector-specific :meth:`_process_events`
+        implementation which turns raw selector events into ready
+        callbacks.
+
+        .. versionadded:: 3.16
+        """
         self._process_events(event_list)
-        # Needed to break cycles when an exception occurs.
-        event_list = None
 
+    def process_ready(self):
+        """Process expired timers and execute ready callbacks.
+
+        Moves scheduled callbacks whose deadline has passed into the
+        ready queue, then runs all callbacks that were ready at call
+        time.  Callbacks enqueued *by* running callbacks are left for
+        the next iteration.
+
+        .. versionadded:: 3.16
+        """
         # Handle 'later' callbacks that are ready.
         now = self.time()
         # Ensure that `end_time` is strictly increasing
@@ -2073,6 +2101,18 @@ def _run_once(self):
                     self._current_handle = None
             else:
                 handle._run()
+
+    def _run_once(self):
+        """Run one full iteration of the event loop.
+
+        This calls all currently ready callbacks, polls for I/O,
+        schedules the resulting callbacks, and finally schedules
+        'call_later' callbacks.
+        """
+        event_list = self.poll_events()
+        self.process_events(event_list)
+        event_list = None       # Needed to break cycles on exception.
+        self.process_ready()
         handle = None  # Needed to break cycles when an exception occurs.
 
     def _set_coroutine_origin_tracking(self, enabled):
diff --git a/Lib/asyncio/guest.py b/Lib/asyncio/guest.py
new file mode 100644
index 00000000000000..bcc9724b5f31cc
--- /dev/null
+++ b/Lib/asyncio/guest.py
@@ -0,0 +1,355 @@
+"""Support for running asyncio as a guest inside another event loop.
+
+This module provides start_guest_run(), which allows asyncio to run
+cooperatively inside a host event loop such as a GUI toolkit's main loop.
+The host loop stays in control of its thread while asyncio tasks execute
+through a dual-thread architecture:
+
+  Host thread:  process_events() + process_ready() -> hand token to I/O thread
+  I/O thread:   wait for token -> poll_events() -> hand events to host
+
+A single "token" (semaphore permit) ping-pongs between the two threads,
+so exactly one of them touches the event loop at any moment.
+
+Inspired by Trio's guest mode (trio.lowlevel.start_guest_run).  The
+asyncio-guest project, which this implementation grew out of, has
+runnable examples for Tkinter, Qt, GTK, pygame, Win32 and Tornado
+hosts:
+
+    https://github.com/congzhangzh/asyncio-guest
+"""
+
+__all__ = ('start_guest_run',)
+
+import signal
+import threading
+from functools import partial
+
+from . import constants
+from . import events
+from . import runners
+from . import tasks
+from .log import logger
+
+# Stop callbacks of unfinished guest runs, so interpreter shutdown can
+# unblock and join their (non-daemon) I/O threads.
+_lock = threading.Lock()
+_stoppers = set()
+_shutting_down = False
+_atexit_registered = False
+
+
+def _python_exit():
+    """Wake and join the I/O threads of all unfinished guest runs.
+
+    Registered via threading._register_atexit() so it runs before the
+    interpreter joins non-daemon threads (the concurrent.futures
+    pattern).  Each run is abandoned: pending tasks are not cancelled
+    and done_callback is not called.
+    """
+    global _shutting_down
+    with _lock:
+        _shutting_down = True
+        stoppers = list(_stoppers)
+    for stop in stoppers:
+        stop()
+
+
+def _ensure_atexit():
+    global _atexit_registered
+    with _lock:
+        if _shutting_down:
+            raise RuntimeError(
+                'cannot start an asyncio guest run at interpreter shutdown')
+        if not _atexit_registered:
+            threading._register_atexit(_python_exit)
+            _atexit_registered = True
+
+
+def _save_wakeup_fd():
+    """Return the current signal wakeup fd, or None if it cannot be read.
+
+    There is no getter for the wakeup fd, so briefly swapping it out is
+    the only way to read it.  A signal arriving between the two calls
+    loses its wakeup byte (its Python-level handler still runs); the
+    window is a few instructions wide.
+    """
+    if (not hasattr(signal, 'set_wakeup_fd')
+            or threading.current_thread() is not threading.main_thread()):
+        return None
+    fd = signal.set_wakeup_fd(-1)
+    if fd != -1:
+        signal.set_wakeup_fd(fd)
+    return fd
+
+
+def _restore_wakeup_fd(fd):
+    if fd is not None:
+        signal.set_wakeup_fd(fd)
+
+
+def start_guest_run(async_fn, *args,
+                    run_sync_soon_threadsafe,
+                    done_callback):
+    """Run async_fn(*args) as a guest inside another event loop.
+
+    Must be called from the host event loop's thread.  The host loop
+    (e.g. Tkinter's mainloop) remains in control of that thread; asyncio
+    I/O polling runs in a background non-daemon thread that is joined
+    when the run finishes.  Returns the Task wrapping *async_fn*; to
+    cancel it from the host, use loop.call_soon_threadsafe(task.cancel)
+    so that the I/O thread is woken from its selector wait.
+
+    run_sync_soon_threadsafe is a callable that schedules a
+    zero-argument callable to run on the host loop's thread.  It must be
+    thread-safe, must not block, and must not raise; it need not
+    preserve ordering.
+
+    done_callback is called on the host thread after the run has fully
+    finished: remaining tasks cancelled, asynchronous generators and the
+    default executor shut down, and the loop closed (the same cleanup as
+    asyncio.run()).  It receives the completed Task as its sole
+    argument.
+
+    For the whole run the guest loop is the thread's running loop:
+    get_running_loop() works inside guest tasks, loop.is_running() is
+    true, and starting another loop on the thread raises RuntimeError.
+    Signal handling stays with the host: the guest loop never touches
+    signal.set_wakeup_fd(), and loop.add_signal_handler() raises
+    RuntimeError.  The host may forward signals into the loop with
+    loop.call_soon_threadsafe().
+
+    If the interpreter exits while the run is unfinished, the run is
+    abandoned: the I/O thread is woken and joined, pending tasks are not
+    cancelled, and done_callback is not called.
+    """
+    _ensure_atexit()
+
+    # Create the loop without letting it capture the signal wakeup fd:
+    # on Windows, BaseProactorEventLoop.__init__ installs its self-pipe
+    # as the wakeup fd when on the main thread, which would silently
+    # break the host's signal handling.
+    host_wakeup_fd = _save_wakeup_fd()
+    loop = events.new_event_loop()
+    _restore_wakeup_fd(host_wakeup_fd)
+
+    # The host owns signal handling (see the docstring).  Never reset:
+    # 'finally' blocks of tasks cancelled during the final cleanup must
+    # not be able to install signal handlers either.
+    loop._guest_mode = True
+
+    def _close_loop():
+        # BaseProactorEventLoop.close() resets the wakeup fd to -1 on
+        # the main thread; preserve the host's fd across it.
+        fd = _save_wakeup_fd()
+        try:
+            loop.close()
+        finally:
+            _restore_wakeup_fd(fd)
+
+    # Mark the loop as running for the whole guest run.  On Windows this
+    # also starts the proactor's self-reading loop, so that
+    # call_soon_threadsafe() can wake the I/O thread's poll.
+    try:
+        loop._run_forever_setup()
+    except BaseException:
+        _close_loop()
+        raise
+
+    shutdown = threading.Event()
+    wakeup = threading.Semaphore(0)
+    finished = False
+    cleaned_up = False
+
+    try:
+        main_task = loop.create_task(async_fn(*args))
+    except BaseException:
+        loop._run_forever_cleanup()
+        _close_loop()
+        raise
+
+    # -- helpers (host thread unless noted) ----------------------------
+
+    def _cleanup_running_state():
+        nonlocal cleaned_up
+        if cleaned_up:
+            return
+        cleaned_up = True
+        loop._run_forever_cleanup()
+
+    def _join_backend():
+        # By the token invariant the I/O thread is parked in
+        # wakeup.acquire() here, never in the selector; the
+        # _write_to_self() wake-up is defensive insurance.
+        shutdown.set()
+        wakeup.release()
+        try:
+            loop._write_to_self()
+        except Exception:
+            pass
+        if backend_thread.ident is not None:
+            backend_thread.join()
+
+    def _deliver(task, *, graceful):
+        with _lock:
+            _stoppers.discard(_stop_at_exit)
+        try:
+            if graceful:
+                # Same cleanup as asyncio.run().
+                runners._cancel_all_tasks(loop)
+                loop.run_until_complete(loop.shutdown_asyncgens())
+                loop.run_until_complete(
+                    loop.shutdown_default_executor(
+                        constants.THREAD_JOIN_TIMEOUT))
+        finally:
+            _close_loop()
+            done_callback(task)
+
+    def _finish(task):
+        # Always scheduled via run_sync_soon_threadsafe, never called
+        # inline from a loop callback: the cleanup below drives the loop
+        # with run_until_complete(), which would re-enter the
+        # process_ready() iteration such a callback runs in.
+        nonlocal finished
+        if finished:
+            return
+        finished = True
+        _join_backend()
+        _cleanup_running_state()
+        _deliver(task, graceful=True)
+
+    def _abort(exc):
+        # The I/O thread died; the loop can no longer be driven through
+        # the guest handshake.  Unwind the main task directly.
+        nonlocal finished
+        if finished:
+            return
+        finished = True
+        _join_backend()
+        _cleanup_running_state()
+        if not main_task.done():
+            main_task.cancel(
+                msg=f'asyncio guest I/O thread failed: {exc!r}')
+            try:
+                loop.run_until_complete(
+                    tasks.gather(main_task, return_exceptions=True))
+            except Exception:
+                # The loop itself is broken (e.g. the selector raised);
+                # abandon any pending work.
+                pass
+        loop.call_exception_handler({
+            'message': 'asyncio guest I/O thread failed',
+            'exception': exc,
+            'task': main_task,
+        })
+        _deliver(main_task, graceful=False)
+
+    def _process_on_host(event_list):
+        """Process one batch of asyncio work on the host thread."""
+        if shutdown.is_set() or loop.is_closed():
+            return
+        try:
+            loop.process_events(event_list)
+            loop.process_ready()
+        except BaseException:
+            # Internal loop failure (user callback exceptions are routed
+            # to the exception handler by Handle._run).  Unblock the I/O
+            # thread so it can exit and be joined.
+            shutdown.set()
+            wakeup.release()
+            raise
+        # Hand the polling token back to the I/O thread.  Exactly one
+        # token circulates: while the host runs a batch, the I/O thread
+        # is parked in wakeup.acquire(), so it can never poll (or touch
+        # the loop at all) concurrently with this function.
+        if not shutdown.is_set():
+            wakeup.release()
+
+    def _on_task_done(task):
+        # Runs inside process_ready() while the host holds the token,
+        # so the I/O thread is parked in wakeup.acquire(): the released
+        # token wakes it, it observes 'shutdown', and exits without
+        # re-entering the selector.
+        shutdown.set()
+        wakeup.release()
+        try:
+            run_sync_soon_threadsafe(partial(_finish, task))
+        except Exception as exc:
+            loop.call_exception_handler({
+                'message': ('asyncio guest run could not schedule its '
+                            'completion callback on the host'),
+                'exception': exc,
+                'task': task,
+            })
+
+    def _backend():
+        """I/O thread: wait for the token, poll, hand events to the host."""
+        try:
+            while True:
+                wakeup.acquire()
+                if shutdown.is_set():
+                    return
+                event_list = loop.poll_events()
+                if shutdown.is_set():
+                    # Interpreter exit woke the selector; the host may
+                    # be gone, so do not call into it.
+                    return
+                run_sync_soon_threadsafe(
+                    partial(_process_on_host, event_list))
+        except Exception as exc:
+            if shutdown.is_set():
+                return
+            shutdown.set()
+            try:
+                run_sync_soon_threadsafe(partial(_abort, exc))
+            except Exception:
+                logger.error(
+                    'asyncio guest run abandoned: host is unreachable '
+                    'after an I/O thread failure', exc_info=True)
+
+    def _stop_at_exit():
+        # Interpreter-exit hook: unblock and join the I/O thread so
+        # shutdown does not hang on a non-daemon thread.  The run is
+        # abandoned -- no cancellation, no callbacks, no loop.close().
+        shutdown.set()
+        wakeup.release()
+        try:
+            loop._write_to_self()
+        except Exception:
+            pass
+        if backend_thread.ident is not None:
+            backend_thread.join()
+
+    # -- start ---------------------------------------------------------
+
+    main_task.add_done_callback(_on_task_done)
+
+    backend_thread = threading.Thread(
+        target=_backend, name='asyncio-guest-io')
+
+    with _lock:
+        shutting_down = _shutting_down
+        if not shutting_down:
+            _stoppers.add(_stop_at_exit)
+    if shutting_down:
+        loop._run_forever_cleanup()
+        _close_loop()
+        raise RuntimeError(
+            'cannot start an asyncio guest run at interpreter shutdown')
+
+    try:
+        # Process the callbacks enqueued by create_task(), then let the
+        # I/O thread take over polling.  The thread is started even if
+        # the task already finished: it consumes the shutdown token and
+        # exits, keeping the join logic uniform.
+        _process_on_host([])
+        backend_thread.start()
+    except BaseException:
+        with _lock:
+            _stoppers.discard(_stop_at_exit)
+        shutdown.set()
+        _cleanup_running_state()
+        _close_loop()
+        raise
+
+    return main_task
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index 3a66cee93da4f5..83aa9a5848d448 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -101,6 +101,11 @@ def add_signal_handler(self, sig, callback, *args):
                             "with add_signal_handler()")
         self._check_signal(sig)
         self._check_closed()
+        if self._guest_mode:
+            raise RuntimeError(
+                "add_signal_handler() is not supported in asyncio guest "
+                "mode; the host event loop owns signal handling -- "
+                "forward signals to the loop with call_soon_threadsafe()")
         try:
             # set_wakeup_fd() raises ValueError if this is not the
             # main thread.  By calling it early we ensure that an
@@ -150,6 +155,10 @@ def remove_signal_handler(self, sig):
         Return True if a signal handler was removed, False if not.
         """
         self._check_signal(sig)
+        if self._guest_mode:
+            raise RuntimeError(
+                "remove_signal_handler() is not supported in asyncio "
+                "guest mode; the host event loop owns signal handling")
         try:
             del self._signal_handlers[sig]
         except KeyError:
diff --git a/Lib/test/test_asyncio/test_guest.py b/Lib/test/test_asyncio/test_guest.py
new file mode 100644
index 00000000000000..01bcdbd4567646
--- /dev/null
+++ b/Lib/test/test_asyncio/test_guest.py
@@ -0,0 +1,434 @@
+"""Tests for asyncio.start_guest_run()."""
+
+import asyncio
+import queue
+import signal
+import socket
+import sys
+import threading
+import time
+import unittest
+from test.support import threading_helper
+from test.support.script_helper import assert_python_ok
+
+threading_helper.requires_working_threading(module=True)
+
+
+def tearDownModule():
+    asyncio.set_event_loop(None)
+
+
+class MockHost:
+    """A minimal host event loop that uses a thread-safe queue.
+
+    Simulates a GUI toolkit main loop without any actual GUI dependency.
+    Callbacks are collected in a queue and drained by :meth:`run`.
+    """
+
+    def __init__(self):
+        self._queue = queue.Queue()
+        self._done = threading.Event()
+        self._task = None
+
+    def run_sync_soon_threadsafe(self, fn):
+        self._queue.put(fn)
+
+    def done_callback(self, task):
+        self._task = task
+        self._done.set()
+
+    def run(self, timeout=10.0):
+        """Drain callbacks until *done_callback* fires or *timeout* expires."""
+        deadline = time.monotonic() + timeout
+        while not self._done.is_set():
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise TimeoutError("MockHost.run() timed out")
+            try:
+                fn = self._queue.get(timeout=min(remaining, 0.05))
+                fn()
+            except queue.Empty:
+                pass
+        # Drain any trailing callbacks.
+        while True:
+            try:
+                fn = self._queue.get_nowait()
+                fn()
+            except queue.Empty:
+                break
+        return self._task
+
+
+class GuestTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self._thread_key = threading_helper.threading_setup()
+
+    def tearDown(self):
+        threading_helper.threading_cleanup(*self._thread_key)
+
+    def _run_guest(self, async_fn, *args, timeout=10.0):
+        """Helper: run *async_fn* in guest mode and return the completed task."""
+        host = MockHost()
+        asyncio.start_guest_run(
+            async_fn, *args,
+            run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
+            done_callback=host.done_callback,
+        )
+        return host.run(timeout=timeout)
+
+
+class TestGuestRun(GuestTestCase):
+    """Test asyncio.start_guest_run with a mock host loop."""
+
+    # -- basic lifecycle -----------------------------------------------
+
+    def test_simple_return(self):
+        async def coro():
+            return 42
+
+        task = self._run_guest(coro)
+        self.assertTrue(task.done())
+        self.assertEqual(task.result(), 42)
+
+    def test_return_none(self):
+        async def coro():
+            pass
+
+        task = self._run_guest(coro)
+        self.assertIsNone(task.result())
+
+    def test_with_args(self):
+        async def add(a, b):
+            return a + b
+
+        task = self._run_guest(add, 3, 7)
+        self.assertEqual(task.result(), 10)
+
+    def test_early_sync_completion(self):
+        # The task can already be done when the I/O thread starts.
+        async def coro():
+            return 'early'
+
+        host = MockHost()
+        task = asyncio.start_guest_run(
+            coro,
+            run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
+            done_callback=host.done_callback,
+        )
+        self.assertIs(host.run(), task)
+        self.assertEqual(task.result(), 'early')
+
+    # -- exception propagation -----------------------------------------
+
+    def test_exception(self):
+        async def coro():
+            raise ValueError("boom")
+
+        task = self._run_guest(coro)
+        self.assertTrue(task.done())
+        with self.assertRaises(ValueError) as cm:
+            task.result()
+        self.assertEqual(str(cm.exception), "boom")
+
+    # -- cancellation --------------------------------------------------
+
+    def test_cancel_from_host(self):
+        started = threading.Event()
+
+        async def coro():
+            started.set()
+            await asyncio.sleep(3600)
+
+        host = MockHost()
+        task = asyncio.start_guest_run(
+            coro,
+            run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
+            done_callback=host.done_callback,
+        )
+        # Wait for the coroutine to start, then cancel.
+        # Use call_soon_threadsafe to wake the I/O thread's selector.
+        started.wait(timeout=5)
+        loop = task.get_loop()
+        loop.call_soon_threadsafe(task.cancel)
+        host.run(timeout=5)
+        self.assertTrue(task.cancelled())
+
+    # -- asyncio primitives work inside guest --------------------------
+
+    def test_sleep(self):
+        async def coro():
+            loop = asyncio.get_running_loop()
+            t0 = loop.time()
+            await asyncio.sleep(0.1)
+            return loop.time() - t0
+
+        task = self._run_guest(coro)
+        self.assertGreaterEqual(task.result(), 0.05)
+
+    def test_create_task(self):
+        async def helper():
+            await asyncio.sleep(0.01)
+            return "helper"
+
+        async def coro():
+            t = asyncio.ensure_future(helper())
+            return await t
+
+        task = self._run_guest(coro)
+        self.assertEqual(task.result(), "helper")
+
+    def test_gather(self):
+        async def sleeper(n):
+            await asyncio.sleep(0.01 * n)
+            return n
+
+        async def coro():
+            return await asyncio.gather(sleeper(1), sleeper(2), sleeper(3))
+
+        task = self._run_guest(coro)
+        self.assertEqual(task.result(), [1, 2, 3])
+
+    def test_call_later(self):
+        async def coro():
+            loop = asyncio.get_running_loop()
+            fut = loop.create_future()
+            loop.call_later(0.05, fut.set_result, "later")
+            return await fut
+
+        task = self._run_guest(coro)
+        self.assertEqual(task.result(), "later")
+
+    def test_call_soon_threadsafe(self):
+        timer = None
+
+        async def coro():
+            nonlocal timer
+            loop = asyncio.get_running_loop()
+            fut = loop.create_future()
+
+            def setter():
+                loop.call_soon_threadsafe(fut.set_result, "safe")
+            timer = threading.Timer(0.05, setter)
+            timer.start()
+            return await fut
+
+        task = self._run_guest(coro)
+        self.assertEqual(task.result(), "safe")
+        timer.join()
+
+    # -- thread lifecycle ----------------------------------------------
+
+    def test_io_thread_nondaemon_and_joined(self):
+        seen = {}
+
+        async def coro():
+            # The I/O thread is started after the initial batch; a sleep
+            # guarantees it is up and polling by the time we look.
+            await asyncio.sleep(0.01)
+            for thread in threading.enumerate():
+                if thread.name == 'asyncio-guest-io':
+                    seen['thread'] = thread
+
+        task = self._run_guest(coro)
+        self.assertIsNone(task.exception())
+        self.assertIn('thread', seen)
+        self.assertFalse(seen['thread'].daemon)
+        self.assertFalse(seen['thread'].is_alive())
+
+    def test_interpreter_exit_with_pending_run(self):
+        # Exiting with an unfinished guest run must not hang: the atexit
+        # hook wakes the non-daemon I/O thread out of its selector wait
+        # and joins it.
+        code = (
+            'import asyncio, collections\n'
+            'q = collections.deque()\n'
+            'async def coro():\n'
+            '    await asyncio.sleep(3600)\n'
+            'asyncio.start_guest_run(\n'
+            '    coro,\n'
+            '    run_sync_soon_threadsafe=q.append,\n'
+            '    done_callback=lambda task: None,\n'
+            ')\n'
+        )
+        assert_python_ok('-c', code)
+
+    # -- running-loop semantics ----------------------------------------
+
+    def test_is_running_inside(self):
+        async def coro():
+            return asyncio.get_running_loop().is_running()
+
+        task = self._run_guest(coro)
+        self.assertTrue(task.result())
+
+    def test_nested_run_raises(self):
+        test = self
+
+        async def coro():
+            loop = asyncio.get_running_loop()
+            inner = asyncio.sleep(0)
+            try:
+                with test.assertRaises(RuntimeError):
+                    loop.run_until_complete(inner)
+            finally:
+                inner.close()
+            inner = asyncio.sleep(0)
+            try:
+                with test.assertRaises(RuntimeError):
+                    asyncio.run(inner)
+            finally:
+                inner.close()
+
+        task = self._run_guest(coro)
+        self.assertIsNone(task.exception())
+
+    def test_state_restored_after_run(self):
+        old_hooks = sys.get_asyncgen_hooks()
+
+        async def coro():
+            pass
+
+        task = self._run_guest(coro)
+        self.assertEqual(sys.get_asyncgen_hooks(), old_hooks)
+        self.assertIsNone(asyncio._get_running_loop())
+        self.assertFalse(task.get_loop().is_running())
+
+    # -- signal handling -----------------------------------------------
+
+    @unittest.skipUnless(hasattr(signal, 'SIGUSR1'),
+                         'requires UNIX signal handling')
+    def test_add_signal_handler_raises(self):
+        async def coro():
+            loop = asyncio.get_running_loop()
+            loop.add_signal_handler(signal.SIGUSR1, lambda: None)
+
+        task = self._run_guest(coro)
+        with self.assertRaisesRegex(RuntimeError, 'guest mode'):
+            task.result()
+
+    @unittest.skipUnless(hasattr(signal, 'SIGUSR1'),
+                         'requires UNIX signal handling')
+    def test_remove_signal_handler_raises(self):
+        async def coro():
+            loop = asyncio.get_running_loop()
+            loop.remove_signal_handler(signal.SIGUSR1)
+
+        task = self._run_guest(coro)
+        with self.assertRaisesRegex(RuntimeError, 'guest mode'):
+            task.result()
+
+    @unittest.skipUnless(hasattr(signal, 'set_wakeup_fd'),
+                         'requires signal.set_wakeup_fd')
+    def test_wakeup_fd_preserved(self):
+        if threading.current_thread() is not threading.main_thread():
+            self.skipTest('requires the main thread')
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(rsock.close)
+        self.addCleanup(wsock.close)
+        wsock.setblocking(False)
+        old_fd = signal.set_wakeup_fd(wsock.fileno())
+        self.addCleanup(signal.set_wakeup_fd, old_fd)
+
+        async def coro():
+            await asyncio.sleep(0.01)
+
+        self._run_guest(coro)
+
+        fd = signal.set_wakeup_fd(-1)
+        if fd != -1:
+            signal.set_wakeup_fd(fd)
+        self.assertEqual(fd, wsock.fileno())
+
+    # -- final cleanup matches asyncio.run() ---------------------------
+
+    def test_background_task_cancelled_on_finish(self):
+        state = {}
+
+        async def background():
+            await asyncio.sleep(3600)
+
+        async def coro():
+            state['bg'] = asyncio.get_running_loop().create_task(background())
+            await asyncio.sleep(0.01)
+
+        task = self._run_guest(coro)
+        self.assertIsNone(task.exception())
+        self.assertTrue(state['bg'].cancelled())
+
+    def test_abandoned_asyncgen_finalized(self):
+        finalized = False
+        holder = []
+
+        async def agen():
+            nonlocal finalized
+            try:
+                yield 1
+            finally:
+                finalized = True
+
+        async def coro():
+            it = agen()
+            holder.append(it)  # keep it alive until shutdown_asyncgens()
+            await anext(it)
+
+        task = self._run_guest(coro)
+        self.assertIsNone(task.exception())
+        self.assertTrue(finalized)
+
+    def test_loop_closed_in_done_callback(self):
+        # Cleanup (cancel remaining tasks, close the loop) happens
+        # before done_callback, like asyncio.run().
+        seen = {}
+        host = MockHost()
+        original = host.done_callback
+
+        def done_callback(task):
+            seen['closed'] = task.get_loop().is_closed()
+            original(task)
+
+        async def coro():
+            pass
+
+        asyncio.start_guest_run(
+            coro,
+            run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
+            done_callback=done_callback,
+        )
+        host.run()
+        self.assertTrue(seen['closed'])
+
+    def test_loop_closed_after_run(self):
+        async def coro():
+            pass
+
+        task = self._run_guest(coro)
+        self.assertTrue(task.get_loop().is_closed())
+
+
+class TestBaseEventLoopDecomposition(GuestTestCase):
+    """Verify that poll_events / process_events / process_ready exist
+    and compose correctly (i.e. _run_once still works)."""
+
+    def test_methods_exist(self):
+        loop = asyncio.new_event_loop()
+        try:
+            self.assertTrue(hasattr(loop, 'poll_events'))
+            self.assertTrue(hasattr(loop, 'process_events'))
+            self.assertTrue(hasattr(loop, 'process_ready'))
+        finally:
+            loop.close()
+
+    def test_run_once_still_works(self):
+        """asyncio.run() exercises _run_once(); ensure it still functions
+        after the refactor."""
+        async def coro():
+            await asyncio.sleep(0)
+            return "ok"
+
+        result = asyncio.run(coro())
+        self.assertEqual(result, "ok")
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst
new file mode 100644
index 00000000000000..1023249cd5ea47
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst
@@ -0,0 +1,12 @@
+Add :func:`asyncio.start_guest_run` to allow asyncio to run cooperatively
+inside a host event loop (e.g. Tkinter, Qt, GTK). The host loop retains
+control of its thread while asyncio I/O polling runs in a background
+non-daemon thread that is joined when the run finishes. In guest mode the
+host owns signal handling: the loop never touches
+:func:`signal.set_wakeup_fd` and :meth:`loop.add_signal_handler
+<asyncio.loop.add_signal_handler>` raises :exc:`RuntimeError`. Also adds
+three low-level event loop methods -- :meth:`loop.poll_events
+<asyncio.loop.poll_events>`, :meth:`loop.process_events
+<asyncio.loop.process_events>`, and :meth:`loop.process_ready
+<asyncio.loop.process_ready>` -- that decompose a single iteration of the
+event loop into independently callable steps.