From f843ee902d7afebca730ff110d0efefe9f4514d5 Mon Sep 17 00:00:00 2001
From: A5rocks <git@helvetica.moe>
Date: Fri, 28 Mar 2025 09:32:53 +0900
Subject: [PATCH 1/7] Copy changes from
 https://github.com/python-trio/trio/pull/1537

Co-authored-by: Joshua Oreman <4316136+oremanj@users.noreply.github.com>
---
 docs/source/reference-lowlevel.rst       | 21 +++++--
 newsfragments/2649.removal.rst           |  4 ++
 newsfragments/733.breaking.rst           | 42 ++++++++++++++
 src/trio/_channel.py                     |  6 +-
 src/trio/_core/_exceptions.py            | 16 ++++++
 src/trio/_core/_generated_io_kqueue.py   |  4 +-
 src/trio/_core/_io_epoll.py              |  4 +-
 src/trio/_core/_io_kqueue.py             |  8 +--
 src/trio/_core/_io_windows.py            | 16 +++---
 src/trio/_core/_parking_lot.py           |  2 +-
 src/trio/_core/_run.py                   | 70 ++++++------------------
 src/trio/_core/_tests/test_asyncgen.py   |  2 +-
 src/trio/_core/_tests/test_guest_mode.py |  3 +-
 src/trio/_core/_tests/test_ki.py         | 64 +++++++++-------------
 src/trio/_core/_tests/test_run.py        | 14 ++---
 src/trio/_core/_traps.py                 | 59 ++++++++++----------
 src/trio/_subprocess_platform/kqueue.py  |  2 +-
 src/trio/_sync.py                        |  3 +-
 src/trio/_tests/test_exports.py          |  8 +++
 src/trio/_threads.py                     | 16 +++---
 src/trio/_tools/gen_exports.py           |  2 +-
 src/trio/lowlevel.py                     | 15 ++++-
 22 files changed, 212 insertions(+), 169 deletions(-)
 create mode 100644 newsfragments/2649.removal.rst
 create mode 100644 newsfragments/733.breaking.rst

diff --git a/docs/source/reference-lowlevel.rst b/docs/source/reference-lowlevel.rst
index 82bd8537d9..15289ff855 100644
--- a/docs/source/reference-lowlevel.rst
+++ b/docs/source/reference-lowlevel.rst
@@ -344,6 +344,8 @@ Spawning threads
 .. autofunction:: start_thread_soon
 
 
+.. _ki-handling:
+
 Safer KeyboardInterrupt handling
 ================================
 
@@ -355,10 +357,21 @@ correctness invariants. On the other, if the user accidentally writes
 an infinite loop, we do want to be able to break out of that. Our
 solution is to install a default signal handler which checks whether
 it's safe to raise :exc:`KeyboardInterrupt` at the place where the
-signal is received. If so, then we do; otherwise, we schedule a
-:exc:`KeyboardInterrupt` to be delivered to the main task at the next
-available opportunity (similar to how :exc:`~trio.Cancelled` is
-delivered).
+signal is received. If so, then we do. Otherwise, we cancel all tasks
+and add `KeyboardInterrupt` as the result of :func:`trio.run`.
+
+.. note:: This behavior means it's not a good idea to try to catch
+   `KeyboardInterrupt` within a Trio task. Most Trio
+   programs are I/O-bound, so most interrupts will be received while
+   no task is running (because Trio is waiting for I/O). There's no
+   task that should obviously receive the interrupt in such cases, so
+   Trio doesn't raise it within a task at all: every task gets cancelled,
+   then `KeyboardInterrupt` is raised once that's complete.
+
+   If you want to handle Ctrl+C by doing something other than "cancel
+   all tasks", then you should use :func:`~trio.open_signal_receiver` to
+   install a handler for `signal.SIGINT`. If you do that, then Ctrl+C will
+   go to your handler, and it can do whatever it wants.
 
 So that's great, but – how do we know whether we're in one of the
 sensitive parts of the program or not?
diff --git a/newsfragments/2649.removal.rst b/newsfragments/2649.removal.rst
new file mode 100644
index 0000000000..0c4aee38a7
--- /dev/null
+++ b/newsfragments/2649.removal.rst
@@ -0,0 +1,4 @@
+The abort function passed to :func:`~trio.lowlevel.wait_task_rescheduled`
+now directly takes as argument the cancellation exception that should be
+raised after a successful asynchronous cancellation. Previously, it took
+a callable that would raise the exception when called.
diff --git a/newsfragments/733.breaking.rst b/newsfragments/733.breaking.rst
new file mode 100644
index 0000000000..cf87925ce4
--- /dev/null
+++ b/newsfragments/733.breaking.rst
@@ -0,0 +1,42 @@
+:ref:`Sometimes <ki-handling>`, a Trio program receives an interrupt
+signal (Ctrl+C) at a time when Python's default response (raising
+`KeyboardInterrupt` immediately) might corrupt Trio's internal
+state. Previously, Trio would handle this situation by raising the
+`KeyboardInterrupt` at the next :ref:`checkpoint <checkpoints>` executed
+by the main task (the one running the function you passed to :func:`trio.run`).
+This was responsible for a lot of internal complexity and sometimes led to
+surprising behavior.
+
+With this release, such a "deferred" `KeyboardInterrupt` is handled in a
+different way: Trio will first cancel all running tasks, then raise
+`KeyboardInterrupt` directly out of the call to :func:`trio.run`.
+The difference is relevant if you have code that tries to catch
+`KeyboardInterrupt` within Trio. This was never entirely robust, but it
+previously might have worked in many cases, whereas now it will never
+catch the interrupt.
+
+An example of code that mostly worked on previous releases, but won't
+work on this release::
+
+    async def main():
+        try:
+            await trio.sleep_forever()
+        except KeyboardInterrupt:
+            print("interrupted")
+    trio.run(main)
+
+The fix is to catch `KeyboardInterrupt` outside Trio::
+
+    async def main():
+        await trio.sleep_forever()
+    try:
+        trio.run(main)
+    except KeyboardInterrupt:
+        print("interrupted")
+
+If that doesn't work for you (because you want to respond to
+`KeyboardInterrupt` by doing something other than cancelling all
+tasks), then you can start a task that uses
+`trio.open_signal_receiver` to receive the interrupt signal ``SIGINT``
+directly and handle it however you wish. Such a task takes precedence
+over Trio's default interrupt handling.
diff --git a/src/trio/_channel.py b/src/trio/_channel.py
index 6410d9120c..244fa9e2b7 100644
--- a/src/trio/_channel.py
+++ b/src/trio/_channel.py
@@ -13,7 +13,7 @@
 import trio
 
 from ._abc import ReceiveChannel, ReceiveType, SendChannel, SendType, T
-from ._core import Abort, RaiseCancelT, Task, enable_ki_protection
+from ._core import Abort, Task, enable_ki_protection
 from ._util import NoPublicConstructor, final, generic_function
 
 if TYPE_CHECKING:
@@ -204,7 +204,7 @@ async def send(self, value: SendType) -> None:
         self._state.send_tasks[task] = value
         task.custom_sleep_data = self
 
-        def abort_fn(_: RaiseCancelT) -> Abort:
+        def abort_fn(_: BaseException) -> Abort:
             self._tasks.remove(task)
             del self._state.send_tasks[task]
             return trio.lowlevel.Abort.SUCCEEDED
@@ -352,7 +352,7 @@ async def receive(self) -> ReceiveType:
         self._state.receive_tasks[task] = None
         task.custom_sleep_data = self
 
-        def abort_fn(_: RaiseCancelT) -> Abort:
+        def abort_fn(_: BaseException) -> Abort:
             self._tasks.remove(task)
             del self._state.receive_tasks[task]
             return trio.lowlevel.Abort.SUCCEEDED
diff --git a/src/trio/_core/_exceptions.py b/src/trio/_core/_exceptions.py
index 4996c18f15..0a6f2b13e9 100644
--- a/src/trio/_core/_exceptions.py
+++ b/src/trio/_core/_exceptions.py
@@ -1,3 +1,6 @@
+from typing import NoReturn
+
+from trio import _deprecate
 from trio._util import NoPublicConstructor, final
 
 
@@ -63,6 +66,19 @@ class Cancelled(BaseException, metaclass=NoPublicConstructor):
     def __str__(self) -> str:
         return "Cancelled"
 
+    def __call__(self) -> NoReturn:
+        # If a Cancelled exception is passed to an old abort_fn that
+        # expects a raise_cancel callback, someone will eventually try
+        # to call the exception instead of raising it. Provide a
+        # deprecation warning and raise it instead.
+        _deprecate.warn_deprecated(
+            "wait_task_rescheduled's abort_fn taking a callback argument",
+            "0.30.0",
+            issue=2649,
+            instead="an exception argument",
+        )
+        raise self
+
 
 class BusyResourceError(Exception):
     """Raised when a task attempts to use a resource that some other task is
diff --git a/src/trio/_core/_generated_io_kqueue.py b/src/trio/_core/_generated_io_kqueue.py
index 556d29e1f2..f942877051 100644
--- a/src/trio/_core/_generated_io_kqueue.py
+++ b/src/trio/_core/_generated_io_kqueue.py
@@ -16,7 +16,7 @@
 
     from .. import _core
     from .._file_io import _HasFileNo
-    from ._traps import Abort, RaiseCancelT
+    from ._traps import Abort
 
 assert not TYPE_CHECKING or sys.platform == "darwin"
 
@@ -59,7 +59,7 @@ def monitor_kevent(
 
 @enable_ki_protection
 async def wait_kevent(
-    ident: int, filter: int, abort_func: Callable[[RaiseCancelT], Abort]
+    ident: int, filter: int, abort_func: Callable[[BaseException], Abort]
 ) -> Abort:
     """TODO: these are implemented, but are currently more of a sketch than
     anything real. See `#26
diff --git a/src/trio/_core/_io_epoll.py b/src/trio/_core/_io_epoll.py
index 5e05f0813f..7287a5be8e 100644
--- a/src/trio/_core/_io_epoll.py
+++ b/src/trio/_core/_io_epoll.py
@@ -16,7 +16,7 @@
 if TYPE_CHECKING:
     from typing_extensions import TypeAlias
 
-    from .._core import Abort, RaiseCancelT
+    from .._core import Abort
     from .._file_io import _HasFileNo
 
 
@@ -303,7 +303,7 @@ async def _epoll_wait(self, fd: int | _HasFileNo, attr_name: str) -> None:
         setattr(waiters, attr_name, _core.current_task())
         self._update_registrations(fd)
 
-        def abort(_: RaiseCancelT) -> Abort:
+        def abort(_: BaseException) -> Abort:
             setattr(waiters, attr_name, None)
             self._update_registrations(fd)
             return _core.Abort.SUCCEEDED
diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py
index 9718c4df80..dd104e97b4 100644
--- a/src/trio/_core/_io_kqueue.py
+++ b/src/trio/_core/_io_kqueue.py
@@ -18,7 +18,7 @@
 
     from typing_extensions import TypeAlias
 
-    from .._core import Abort, RaiseCancelT, Task, UnboundedQueue
+    from .._core import Abort, Task, UnboundedQueue
     from .._file_io import _HasFileNo
 
 assert not TYPE_CHECKING or (sys.platform != "linux" and sys.platform != "win32")
@@ -147,7 +147,7 @@ async def wait_kevent(
         self,
         ident: int,
         filter: int,
-        abort_func: Callable[[RaiseCancelT], Abort],
+        abort_func: Callable[[BaseException], Abort],
     ) -> Abort:
         """TODO: these are implemented, but are currently more of a sketch than
         anything real. See `#26
@@ -160,7 +160,7 @@ async def wait_kevent(
             )
         self._registered[key] = _core.current_task()
 
-        def abort(raise_cancel: RaiseCancelT) -> Abort:
+        def abort(raise_cancel: BaseException) -> Abort:
             r = abort_func(raise_cancel)
             if r is _core.Abort.SUCCEEDED:  # TODO: test this branch
                 del self._registered[key]
@@ -180,7 +180,7 @@ async def _wait_common(
         event = select.kevent(fd, filter, flags)
         self._kqueue.control([event], 0)
 
-        def abort(_: RaiseCancelT) -> Abort:
+        def abort(_: BaseException) -> Abort:
             event = select.kevent(fd, filter, select.KQ_EV_DELETE)
             try:
                 self._kqueue.control([event], 0)
diff --git a/src/trio/_core/_io_windows.py b/src/trio/_core/_io_windows.py
index 148253ab88..cbfae9bfe2 100644
--- a/src/trio/_core/_io_windows.py
+++ b/src/trio/_core/_io_windows.py
@@ -45,7 +45,7 @@
     from typing_extensions import Buffer, TypeAlias
 
     from .._file_io import _HasFileNo
-    from ._traps import Abort, RaiseCancelT
+    from ._traps import Abort
     from ._unbounded_queue import UnboundedQueue
 
 EventResult: TypeAlias = int
@@ -752,7 +752,7 @@ async def _afd_poll(self, sock: _HasFileNo | int, mode: str) -> None:
         # we let it escape.
         self._refresh_afd(base_handle)
 
-        def abort_fn(_: RaiseCancelT) -> Abort:
+        def abort_fn(_: BaseException) -> Abort:
             setattr(waiters, mode, None)
             self._refresh_afd(base_handle)
             return _core.Abort.SUCCEEDED
@@ -864,11 +864,11 @@ async def wait_overlapped(
             )
         task = _core.current_task()
         self._overlapped_waiters[lpOverlapped] = task
-        raise_cancel = None
+        cancel_exc = None
 
-        def abort(raise_cancel_: RaiseCancelT) -> Abort:
-            nonlocal raise_cancel
-            raise_cancel = raise_cancel_
+        def abort(cancel_exc_: BaseException) -> Abort:
+            nonlocal cancel_exc
+            cancel_exc = cancel_exc_
             try:
                 _check(kernel32.CancelIoEx(handle, lpOverlapped))
             except OSError as exc:
@@ -914,8 +914,8 @@ def abort(raise_cancel_: RaiseCancelT) -> Abort:
             # it will produce the right sorts of exceptions
             code = ntdll.RtlNtStatusToDosError(lpOverlappedTyped.Internal)
             if code == ErrorCodes.ERROR_OPERATION_ABORTED:
-                if raise_cancel is not None:
-                    raise_cancel()
+                if cancel_exc is not None:
+                    raise cancel_exc
                 else:
                     # We didn't request this cancellation, so assume
                     # it happened due to the underlying handle being
diff --git a/src/trio/_core/_parking_lot.py b/src/trio/_core/_parking_lot.py
index ddf6276117..efd6e83fc0 100644
--- a/src/trio/_core/_parking_lot.py
+++ b/src/trio/_core/_parking_lot.py
@@ -183,7 +183,7 @@ async def park(self) -> None:
         self._parked[task] = None
         task.custom_sleep_data = self
 
-        def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
+        def abort_fn(_: BaseException) -> _core.Abort:
             del task.custom_sleep_data._parked[task]
             return _core.Abort.SUCCEEDED
 
diff --git a/src/trio/_core/_run.py b/src/trio/_core/_run.py
index 1552870ece..1e31efd806 100644
--- a/src/trio/_core/_run.py
+++ b/src/trio/_core/_run.py
@@ -1216,12 +1216,10 @@ async def _nested_child_finished(
             # If we have a KeyboardInterrupt injected, we want to save it in
             # the nursery's final exceptions list. But if it's just a
             # Cancelled, then we don't -- see gh-1457.
-            def aborted(raise_cancel: _core.RaiseCancelT) -> Abort:
-                exn = capture(raise_cancel).error
+            def aborted(exn: BaseException) -> Abort:
+                # TODO: when do we abort with something other than Cancelled?
                 if not isinstance(exn, Cancelled):
                     self._add_exc(exn)
-                # see test_cancel_scope_exit_doesnt_create_cyclic_garbage
-                del exn  # prevent cyclic garbage creation
                 return Abort.FAILED
 
             self._parent_waiting_in_aexit = True
@@ -1424,7 +1422,7 @@ class Task(metaclass=NoPublicConstructor):  # type: ignore[explicit-any]
     # Tasks start out unscheduled.
     _next_send_fn: Callable[[Any], object] | None = None  # type: ignore[explicit-any]
     _next_send: Outcome[Any] | BaseException | None = None  # type: ignore[explicit-any]
-    _abort_func: Callable[[_core.RaiseCancelT], Abort] | None = None
+    _abort_func: Callable[[BaseException], Abort] | None = None
     custom_sleep_data: Any = None  # type: ignore[explicit-any]
 
     # For introspection and nursery.start()
@@ -1538,24 +1536,24 @@ def _activate_cancel_status(self, cancel_status: CancelStatus | None) -> None:
             if self._cancel_status.effectively_cancelled:
                 self._attempt_delivery_of_any_pending_cancel()
 
-    def _attempt_abort(self, raise_cancel: _core.RaiseCancelT) -> None:
+    def _attempt_abort(self, cancel_exc: BaseException) -> None:
         # Either the abort succeeds, in which case we will reschedule the
         # task, or else it fails, in which case it will worry about
-        # rescheduling itself (hopefully eventually calling reraise to raise
+        # rescheduling itself (hopefully eventually calling raising
         # the given exception, but not necessarily).
 
         # This is only called by the functions immediately below, which both check
         # `self.abort_func is not None`.
         assert self._abort_func is not None, "FATAL INTERNAL ERROR"
 
-        success = self._abort_func(raise_cancel)
+        success = self._abort_func(cancel_exc)
         if type(success) is not Abort:
             raise TrioInternalError("abort function must return Abort enum")
         # We only attempt to abort once per blocking call, regardless of
         # whether we succeeded or failed.
         self._abort_func = None
         if success is Abort.SUCCEEDED:
-            self._runner.reschedule(self, capture(raise_cancel))
+            self._runner.reschedule(self, Error(cancel_exc))
 
     def _attempt_delivery_of_any_pending_cancel(self) -> None:
         if self._abort_func is None:
@@ -1563,21 +1561,7 @@ def _attempt_delivery_of_any_pending_cancel(self) -> None:
         if not self._cancel_status.effectively_cancelled:
             return
 
-        def raise_cancel() -> NoReturn:
-            raise Cancelled._create()
-
-        self._attempt_abort(raise_cancel)
-
-    def _attempt_delivery_of_pending_ki(self) -> None:
-        assert self._runner.ki_pending
-        if self._abort_func is None:
-            return
-
-        def raise_cancel() -> NoReturn:
-            self._runner.ki_pending = False
-            raise KeyboardInterrupt
-
-        self._attempt_abort(raise_cancel)
+        self._attempt_abort(Cancelled._create())
 
 
 ################################################################
@@ -1709,6 +1693,7 @@ class Runner:  # type: ignore[explicit-any]
     system_context: contextvars.Context = attrs.field(kw_only=True)
     main_task: Task | None = None
     main_task_outcome: Outcome[object] | None = None
+    main_task_nursery: Nursery | None = None
 
     entry_queue: EntryQueue = attrs.Factory(EntryQueue)
     trio_token: TrioToken | None = None
@@ -2043,12 +2028,12 @@ async def init(
             # All other system tasks run here:
             async with open_nursery() as self.system_nursery:
                 # Only the main task runs here:
-                async with open_nursery() as main_task_nursery:
+                async with open_nursery() as self.main_task_nursery:
                     try:
                         self.main_task = self.spawn_impl(
                             async_fn,
                             args,
-                            main_task_nursery,
+                            self.main_task_nursery,
                             None,
                         )
                     except BaseException as exc:
@@ -2093,30 +2078,13 @@ def current_trio_token(self) -> TrioToken:
 
     ki_pending: bool = False
 
-    # deliver_ki is broke. Maybe move all the actual logic and state into
-    # RunToken, and we'll only have one instance per runner? But then we can't
-    # have a public constructor. Eh, but current_run_token() returning a
-    # unique object per run feels pretty nice. Maybe let's just go for it. And
-    # keep the class public so people can isinstance() it if they want.
-
     # This gets called from signal context
     def deliver_ki(self) -> None:
         self.ki_pending = True
-        with suppress(RunFinishedError):
-            self.entry_queue.run_sync_soon(self._deliver_ki_cb)
+        assert self.main_task_nursery is not None
 
-    def _deliver_ki_cb(self) -> None:
-        if not self.ki_pending:
-            return
-        # Can't happen because main_task and run_sync_soon_task are created at
-        # the same time -- so even if KI arrives before main_task is created,
-        # we won't get here until afterwards.
-        assert self.main_task is not None
-        if self.main_task_outcome is not None:
-            # We're already in the process of exiting -- leave ki_pending set
-            # and we'll check it again on our way out of run().
-            return
-        self.main_task._attempt_delivery_of_pending_ki()
+        with suppress(RunFinishedError):
+            self.entry_queue.run_sync_soon(self.main_task_nursery.cancel_scope.cancel)
 
     ################
     # Quiescing
@@ -2189,7 +2157,7 @@ async def test_lock_fairness():
         key = (cushion, id(task))
         self.waiting_for_idle[key] = task
 
-        def abort(_: _core.RaiseCancelT) -> Abort:
+        def abort(_: BaseException) -> Abort:
             del self.waiting_for_idle[key]
             return Abort.SUCCEEDED
 
@@ -2775,10 +2743,6 @@ def unrolled_run(
                     elif type(msg) is WaitTaskRescheduled:
                         task._cancel_points += 1
                         task._abort_func = msg.abort_func
-                        # KI is "outside" all cancel scopes, so check for it
-                        # before checking for regular cancellation:
-                        if runner.ki_pending and task is runner.main_task:
-                            task._attempt_delivery_of_pending_ki()
                         task._attempt_delivery_of_any_pending_cancel()
                     elif type(msg) is PermanentlyDetachCoroutineObject:
                         # Pretend the task just exited with the given outcome
@@ -2911,9 +2875,7 @@ async def checkpoint() -> None:
     await cancel_shielded_checkpoint()
     task = current_task()
     task._cancel_points += 1
-    if task._cancel_status.effectively_cancelled or (
-        task is task._runner.main_task and task._runner.ki_pending
-    ):
+    if task._cancel_status.effectively_cancelled:
         with CancelScope(deadline=-inf):
             await _core.wait_task_rescheduled(lambda _: _core.Abort.SUCCEEDED)
 
diff --git a/src/trio/_core/_tests/test_asyncgen.py b/src/trio/_core/_tests/test_asyncgen.py
index 8147a0e57b..29b0aa5e5a 100644
--- a/src/trio/_core/_tests/test_asyncgen.py
+++ b/src/trio/_core/_tests/test_asyncgen.py
@@ -255,7 +255,7 @@ async def step_outside_async_context(aiter_: AsyncGenerator[int, None]) -> None:
     # NB: the strangeness with aiter being an attribute of abort_fn is
     # to make it as easy as possible to ensure we don't hang onto a
     # reference to aiter inside the guts of the run loop.
-    def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
+    def abort_fn(_: BaseException) -> _core.Abort:
         with pytest.raises(StopIteration, match="42"):
             abort_fn.aiter.asend(None).send(None)  # type: ignore[attr-defined]  # Callables don't have attribute "aiter"
         del abort_fn.aiter  # type: ignore[attr-defined]
diff --git a/src/trio/_core/_tests/test_guest_mode.py b/src/trio/_core/_tests/test_guest_mode.py
index 81b7a07d87..3624a9d962 100644
--- a/src/trio/_core/_tests/test_guest_mode.py
+++ b/src/trio/_core/_tests/test_guest_mode.py
@@ -639,7 +639,8 @@ async def trio_main(in_host: InHost) -> None:
 
     with pytest.raises(KeyboardInterrupt) as excinfo:
         trivial_guest_run(trio_main)
-    assert excinfo.value.__context__ is None
+    assert isinstance(excinfo.value.__context__, trio.Cancelled)
+    assert excinfo.value.__context__.__context__ is None
     # Signal handler should be restored properly on exit
     assert signal.getsignal(signal.SIGINT) is signal.default_int_handler
 
diff --git a/src/trio/_core/_tests/test_ki.py b/src/trio/_core/_tests/test_ki.py
index 07a7558720..4b7e329879 100644
--- a/src/trio/_core/_tests/test_ki.py
+++ b/src/trio/_core/_tests/test_ki.py
@@ -36,7 +36,7 @@
         Iterator,
     )
 
-    from ..._core import Abort, RaiseCancelT
+    from ..._core import Abort
 
 
 def ki_self() -> None:
@@ -313,8 +313,8 @@ async def check_unprotected_kill() -> None:
         _core.run(check_unprotected_kill)
     assert record_set == {"s1 ok", "s2 ok", "r1 raise ok"}
 
-    # simulated control-C during raiser, which is *protected*, so the KI gets
-    # delivered to the main task instead
+    # simulated control-C during raiser, which is *protected*, so the run
+    # gets cancelled instead.
     print("check 2")
     record_set = set()
 
@@ -325,9 +325,12 @@ async def check_protected_kill() -> None:
             nursery.start_soon(_core.enable_ki_protection(raiser), "r1", record_set)
             # __aexit__ blocks, and then receives the KI
 
-    # raises inside a nursery, so the KeyboardInterrupt is wrapped in an ExceptionGroup
-    with RaisesGroup(KeyboardInterrupt):
+    # KeyboardInterrupt is inserted from the trio.run
+    with pytest.raises(KeyboardInterrupt) as excinfo:
         _core.run(check_protected_kill)
+
+    # TODO: be consistent about providing Cancelled tree as __context__
+    assert excinfo.value.__context__ is None
     assert record_set == {"s1 ok", "s2 ok", "r1 cancel ok"}
 
     # kill at last moment still raises (run_sync_soon until it raises an
@@ -373,10 +376,11 @@ async def main_1() -> None:
     async def main_2() -> None:
         assert _core.currently_ki_protected()
         ki_self()
-        with pytest.raises(KeyboardInterrupt):
+        with pytest.raises(_core.Cancelled):
             await _core.checkpoint_if_cancelled()
 
-    _core.run(main_2)
+    with pytest.raises(KeyboardInterrupt):
+        _core.run(main_2)
 
     # KI arrives while main task is not abortable, b/c already scheduled
     print("check 6")
@@ -388,10 +392,11 @@ async def main_3() -> None:
         await _core.cancel_shielded_checkpoint()
         await _core.cancel_shielded_checkpoint()
         await _core.cancel_shielded_checkpoint()
-        with pytest.raises(KeyboardInterrupt):
+        with pytest.raises(_core.Cancelled):
             await _core.checkpoint()
 
-    _core.run(main_3)
+    with pytest.raises(KeyboardInterrupt):
+        _core.run(main_3)
 
     # KI arrives while main task is not abortable, b/c refuses to be aborted
     print("check 7")
@@ -402,15 +407,16 @@ async def main_4() -> None:
         ki_self()
         task = _core.current_task()
 
-        def abort(_: RaiseCancelT) -> Abort:
+        def abort(_: BaseException) -> Abort:
             _core.reschedule(task, outcome.Value(1))
             return _core.Abort.FAILED
 
         assert await _core.wait_task_rescheduled(abort) == 1
-        with pytest.raises(KeyboardInterrupt):
+        with pytest.raises(_core.Cancelled):
             await _core.checkpoint()
 
-    _core.run(main_4)
+    with pytest.raises(KeyboardInterrupt):
+        _core.run(main_4)
 
     # KI delivered via slow abort
     print("check 8")
@@ -421,16 +427,16 @@ async def main_5() -> None:
         ki_self()
         task = _core.current_task()
 
-        def abort(raise_cancel: RaiseCancelT) -> Abort:
-            result = outcome.capture(raise_cancel)
-            _core.reschedule(task, result)
+        def abort(cancel_exc: BaseException) -> Abort:
+            _core.reschedule(task, outcome.Error(cancel_exc))
             return _core.Abort.FAILED
 
-        with pytest.raises(KeyboardInterrupt):
+        with pytest.raises(_core.Cancelled):
             assert await _core.wait_task_rescheduled(abort)
         await _core.checkpoint()
 
-    _core.run(main_5)
+    with pytest.raises(KeyboardInterrupt):
+        _core.run(main_5)
 
     # KI arrives just before main task exits, so the run_sync_soon machinery
     # is still functioning and will accept the callback to deliver the KI, but
@@ -457,10 +463,11 @@ async def main_7() -> None:
         # ...but even after the KI, we keep running uninterrupted...
         record_list.append("ok")
         # ...until we hit a checkpoint:
-        with pytest.raises(KeyboardInterrupt):
+        with pytest.raises(_core.Cancelled):
             await sleep(10)
 
-    _core.run(main_7, restrict_keyboard_interrupt_to_checkpoints=True)
+    with pytest.raises(KeyboardInterrupt):
+        _core.run(main_7, restrict_keyboard_interrupt_to_checkpoints=True)
     assert record_list == ["ok"]
     record_list = []
     # Exact same code raises KI early if we leave off the argument, doesn't
@@ -469,25 +476,6 @@ async def main_7() -> None:
         _core.run(main_7)
     assert record_list == []
 
-    # KI arrives while main task is inside a cancelled cancellation scope
-    # the KeyboardInterrupt should take priority
-    print("check 11")
-
-    @_core.enable_ki_protection
-    async def main_8() -> None:
-        assert _core.currently_ki_protected()
-        with _core.CancelScope() as cancel_scope:
-            cancel_scope.cancel()
-            with pytest.raises(_core.Cancelled):
-                await _core.checkpoint()
-            ki_self()
-            with pytest.raises(KeyboardInterrupt):
-                await _core.checkpoint()
-            with pytest.raises(_core.Cancelled):
-                await _core.checkpoint()
-
-    _core.run(main_8)
-
 
 def test_ki_is_good_neighbor() -> None:
     # in the unlikely event someone overwrites our signal handler, we leave
diff --git a/src/trio/_core/_tests/test_run.py b/src/trio/_core/_tests/test_run.py
index 7728a6f3d4..21c6b43a42 100644
--- a/src/trio/_core/_tests/test_run.py
+++ b/src/trio/_core/_tests/test_run.py
@@ -1509,9 +1509,8 @@ async def test_slow_abort_basic() -> None:
         task = _core.current_task()
         token = _core.current_trio_token()
 
-        def slow_abort(raise_cancel: _core.RaiseCancelT) -> _core.Abort:
-            result = outcome.capture(raise_cancel)
-            token.run_sync_soon(_core.reschedule, task, result)
+        def slow_abort(cancel_exc: BaseException) -> _core.Abort:
+            token.run_sync_soon(_core.reschedule, task, outcome.Error(cancel_exc))
             return _core.Abort.FAILED
 
         with pytest.raises(_core.Cancelled):
@@ -1525,10 +1524,9 @@ async def slow_aborter() -> None:
         task = _core.current_task()
         token = _core.current_trio_token()
 
-        def slow_abort(raise_cancel: _core.RaiseCancelT) -> _core.Abort:
+        def slow_abort(cancel_exc: BaseException) -> _core.Abort:
             record.append("abort-called")
-            result = outcome.capture(raise_cancel)
-            token.run_sync_soon(_core.reschedule, task, result)
+            token.run_sync_soon(_core.reschedule, task, outcome.Error(cancel_exc))
             return _core.Abort.FAILED
 
         record.append("sleeping")
@@ -2360,7 +2358,7 @@ async def reattachable_coroutine() -> None:
 
         task = _core.current_task()
 
-        def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:  # pragma: no cover
+        def abort_fn(_: BaseException) -> _core.Abort:  # pragma: no cover
             return _core.Abort.FAILED
 
         got = await _core.temporarily_detach_coroutine_object(abort_fn)
@@ -2404,7 +2402,7 @@ async def reattachable_coroutine() -> None:
         nonlocal task
         task = _core.current_task()
 
-        def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
+        def abort_fn(_: BaseException) -> _core.Abort:
             nonlocal abort_fn_called
             abort_fn_called = True
             return _core.Abort.FAILED
diff --git a/src/trio/_core/_traps.py b/src/trio/_core/_traps.py
index 60f72d1295..ba5b345bfe 100644
--- a/src/trio/_core/_traps.py
+++ b/src/trio/_core/_traps.py
@@ -32,7 +32,7 @@ class CancelShieldedCheckpoint:
 # Not exported in the trio._core namespace, but imported directly by _run.
 @attrs.frozen(slots=False)
 class WaitTaskRescheduled:
-    abort_func: Callable[[RaiseCancelT], Abort]
+    abort_func: Callable[[BaseException], Abort]
 
 
 # Not exported in the trio._core namespace, but imported directly by _run.
@@ -105,7 +105,7 @@ class Abort(enum.Enum):
 # Should always return the type a Task "expects", unless you willfully reschedule it
 # with a bad value.
 async def wait_task_rescheduled(  # type: ignore[explicit-any]
-    abort_func: Callable[[RaiseCancelT], Abort],
+    abort_func: Callable[[BaseException], Abort],
 ) -> Any:
     """Put the current task to sleep, with cancellation support.
 
@@ -137,7 +137,7 @@ async def wait_task_rescheduled(  # type: ignore[explicit-any]
        timeout expiring). When this happens, the ``abort_func`` is called. Its
        interface looks like::
 
-           def abort_func(raise_cancel):
+           def abort_func(exc):
                ...
                return trio.lowlevel.Abort.SUCCEEDED  # or FAILED
 
@@ -151,40 +151,43 @@ def abort_func(raise_cancel):
        task can't be cancelled at this time, and still has to make sure that
        "someone" eventually calls :func:`reschedule`.
 
-       At that point there are again two possibilities. You can simply ignore
-       the cancellation altogether: wait for the operation to complete and
-       then reschedule and continue as normal. (For example, this is what
-       :func:`trio.to_thread.run_sync` does if cancellation is disabled.)
-       The other possibility is that the ``abort_func`` does succeed in
-       cancelling the operation, but for some reason isn't able to report that
-       right away. (Example: on Windows, it's possible to request that an
-       async ("overlapped") I/O operation be cancelled, but this request is
-       *also* asynchronous – you don't find out until later whether the
-       operation was actually cancelled or not.)  To report a delayed
-       cancellation, then you should reschedule the task yourself, and call
-       the ``raise_cancel`` callback passed to ``abort_func`` to raise a
-       :exc:`~trio.Cancelled` (or possibly :exc:`KeyboardInterrupt`) exception
-       into this task. Either of the approaches sketched below can work::
+       At that point there are again two possibilities. You can simply
+       ignore the cancellation altogether: wait for the operation to
+       complete and then reschedule and continue as normal. (For
+       example, this is what :func:`trio.to_thread.run_sync` does if
+       cancellation is disabled.)  The other possibility is that the
+       ``abort_func`` does succeed in cancelling the operation, but
+       for some reason isn't able to report that right away. (Example:
+       on Windows, it's possible to request that an async
+       ("overlapped") I/O operation be cancelled, but this request is
+       *also* asynchronous – you don't find out until later whether
+       the operation was actually cancelled or not.)  To report a
+       delayed cancellation, you should reschedule the task yourself,
+       and cause it to raise the exception ``exc`` that was passed to
+       ``abort_func``. (Currently ``exc`` will always be a
+       `~trio.Cancelled` exception, but we may use this mechanism to
+       raise other exceptions in the future; you should raise whatever
+       you're given.) Either of the approaches sketched below can
+       work::
 
           # Option 1:
-          # Catch the exception from raise_cancel and inject it into the task.
+          # Directly reschedule the task with the provided exception.
           # (This is what Trio does automatically for you if you return
           # Abort.SUCCEEDED.)
-          trio.lowlevel.reschedule(task, outcome.capture(raise_cancel))
+          trio.lowlevel.reschedule(task, outcome.Error(exc))
 
           # Option 2:
           # wait to be woken by "someone", and then decide whether to raise
           # the error from inside the task.
-          outer_raise_cancel = None
-          def abort(inner_raise_cancel):
-              nonlocal outer_raise_cancel
-              outer_raise_cancel = inner_raise_cancel
+          outer_exc = None
+          def abort(inner_exc):
+              nonlocal outer_exc
+              outer_exc = inner_exc
               TRY_TO_CANCEL_OPERATION()
               return trio.lowlevel.Abort.FAILED
           await wait_task_rescheduled(abort)
           if OPERATION_WAS_SUCCESSFULLY_CANCELLED:
-              # raises the error
-              outer_raise_cancel()
+              raise outer_exc
 
        In any case it's guaranteed that we only call the ``abort_func`` at most
        once per call to :func:`wait_task_rescheduled`.
@@ -242,7 +245,7 @@ async def permanently_detach_coroutine_object(
 
 
 async def temporarily_detach_coroutine_object(
-    abort_func: Callable[[RaiseCancelT], Abort],
+    abort_func: Callable[[BaseException], Abort],
 ) -> object:
     """Temporarily detach the current coroutine object from the Trio
     scheduler.
@@ -269,8 +272,8 @@ async def temporarily_detach_coroutine_object(
           detached task directly without going through
           :func:`reattach_detached_coroutine_object`, which would be bad.)
           Your ``abort_func`` should still arrange for whatever the coroutine
-          object is doing to be cancelled, and then reattach to Trio and call
-          the ``raise_cancel`` callback, if possible.
+          object is doing to be cancelled, and then reattach to Trio and raise
+          the exception it received, if possible.
 
     Returns or raises whatever value or exception the new coroutine runner
     uses to resume the coroutine.
diff --git a/src/trio/_subprocess_platform/kqueue.py b/src/trio/_subprocess_platform/kqueue.py
index 2283bb5360..2b257eeea5 100644
--- a/src/trio/_subprocess_platform/kqueue.py
+++ b/src/trio/_subprocess_platform/kqueue.py
@@ -41,7 +41,7 @@ def make_event(flags: int) -> select.kevent:
         # in Chromium it seems we should still keep the check.
         return
 
-    def abort(_: _core.RaiseCancelT) -> _core.Abort:
+    def abort(_: BaseException) -> _core.Abort:
         kqueue.control([make_event(select.KQ_EV_DELETE)], 0)
         return _core.Abort.SUCCEEDED
 
diff --git a/src/trio/_sync.py b/src/trio/_sync.py
index ca373922b0..383dd4168a 100644
--- a/src/trio/_sync.py
+++ b/src/trio/_sync.py
@@ -11,7 +11,6 @@
 from ._core import (
     Abort,
     ParkingLot,
-    RaiseCancelT,
     add_parking_lot_breaker,
     enable_ki_protection,
     remove_parking_lot_breaker,
@@ -95,7 +94,7 @@ async def wait(self) -> None:
             task = _core.current_task()
             self._tasks.add(task)
 
-            def abort_fn(_: RaiseCancelT) -> Abort:
+            def abort_fn(_: BaseException) -> Abort:
                 self._tasks.remove(task)
                 return _core.Abort.SUCCEEDED
 
diff --git a/src/trio/_tests/test_exports.py b/src/trio/_tests/test_exports.py
index 0bf619e3b8..d69fad1104 100644
--- a/src/trio/_tests/test_exports.py
+++ b/src/trio/_tests/test_exports.py
@@ -65,6 +65,7 @@ def test_core_is_properly_reexported() -> None:
     # Each export from _core should be re-exported by exactly one of these
     # three modules:
     sources = [trio, trio.lowlevel, trio.testing]
+    hit_RaiseCancelT = False
     for symbol in dir(_core):
         if symbol.startswith("_"):
             continue
@@ -76,8 +77,15 @@ def test_core_is_properly_reexported() -> None:
             ):
                 found += 1
         print(symbol, found)
+        if symbol == "RaiseCancelT":
+            # deprecated
+            hit_RaiseCancelT = True
+            continue
+
         assert found == 1
 
+    assert hit_RaiseCancelT
+
 
 def class_is_final(cls: type) -> bool:
     """Check if a class cannot be subclassed."""
diff --git a/src/trio/_threads.py b/src/trio/_threads.py
index 394e5b06ac..4eb97b6495 100644
--- a/src/trio/_threads.py
+++ b/src/trio/_threads.py
@@ -31,8 +31,6 @@
 
     from typing_extensions import TypeVarTuple, Unpack
 
-    from trio._core._traps import RaiseCancelT
-
     Ts = TypeVarTuple("Ts")
 
 RetT = TypeVar("RetT")
@@ -44,7 +42,7 @@ class _ParentTaskData(threading.local):
 
     token: TrioToken
     abandon_on_cancel: bool
-    cancel_register: list[RaiseCancelT | None]
+    cancel_register: list[BaseException | None]
     task_register: list[trio.lowlevel.Task | None]
 
 
@@ -357,7 +355,7 @@ async def to_thread_run_sync(
     task_register: list[trio.lowlevel.Task | None] = [trio.lowlevel.current_task()]
     # Holds a reference to the raise_cancel function provided if a cancellation
     # is attempted against this task - or None if no such delivery has happened.
-    cancel_register: list[RaiseCancelT | None] = [None]  # type: ignore[assignment]
+    cancel_register: list[BaseException | None] = [None]
     name = f"trio.to_thread.run_sync-{next(_thread_counter)}"
     placeholder = ThreadPlaceholder(name)
 
@@ -428,9 +426,9 @@ def deliver_worker_fn_result(result: outcome.Outcome[RetT]) -> None:
             limiter.release_on_behalf_of(placeholder)
             raise
 
-        def abort(raise_cancel: RaiseCancelT) -> trio.lowlevel.Abort:
+        def abort(cancel_exc: BaseException) -> trio.lowlevel.Abort:
             # fill so from_thread_check_cancelled can raise
-            cancel_register[0] = raise_cancel
+            cancel_register[0] = cancel_exc
             if abandon_bool:
                 # empty so report_back_in_trio_thread_fn cannot reschedule
                 task_register[0] = None
@@ -484,13 +482,13 @@ def from_thread_check_cancelled() -> None:
        for completeness.
     """
     try:
-        raise_cancel = PARENT_TASK_DATA.cancel_register[0]
+        cancel_exc = PARENT_TASK_DATA.cancel_register[0]
     except AttributeError:
         raise RuntimeError(
             "this thread wasn't created by Trio, can't check for cancellation",
         ) from None
-    if raise_cancel is not None:
-        raise_cancel()
+    if cancel_exc is not None:
+        raise cancel_exc
 
 
 def _send_message_to_trio(
diff --git a/src/trio/_tools/gen_exports.py b/src/trio/_tools/gen_exports.py
index 5b1affe24a..f603a7f3ff 100755
--- a/src/trio/_tools/gen_exports.py
+++ b/src/trio/_tools/gen_exports.py
@@ -382,7 +382,7 @@ def main() -> None:  # pragma: no cover
 
     from .. import _core
     from .._file_io import _HasFileNo
-    from ._traps import Abort, RaiseCancelT
+    from ._traps import Abort
 """
 
 IMPORTS_WINDOWS = """\
diff --git a/src/trio/lowlevel.py b/src/trio/lowlevel.py
index bbeab6af17..7aece2fb4f 100644
--- a/src/trio/lowlevel.py
+++ b/src/trio/lowlevel.py
@@ -16,7 +16,7 @@
     Abort as Abort,
     ParkingLot as ParkingLot,
     ParkingLotStatistics as ParkingLotStatistics,
-    RaiseCancelT as RaiseCancelT,
+    RaiseCancelT as _RaiseCancelT,
     RunStatistics as RunStatistics,
     RunVar as RunVar,
     RunVarToken as RunVarToken,
@@ -84,4 +84,15 @@
             wait_kevent as wait_kevent,
         )
 
-del sys
+from ._deprecate import DeprecatedAttribute, deprecate_attributes
+
+deprecate_attributes(
+    __name__,
+    {
+        "RaiseCancelT": DeprecatedAttribute(
+            _RaiseCancelT, "0.30.0", issue=2649, instead="an exception argument"
+        )
+    },
+)
+
+del sys, deprecate_attributes, DeprecatedAttribute

From 34aab7f7f6386d6dd3173561429c1a56dc25ada8 Mon Sep 17 00:00:00 2001
From: A5rocks <git@helvetica.moe>
Date: Fri, 28 Mar 2025 10:38:15 +0900
Subject: [PATCH 2/7] Make the REPL handle KI per-prompt

---
 src/trio/_repl.py | 43 ++++++++++++++++++++++++++++++++-----------
 1 file changed, 32 insertions(+), 11 deletions(-)

diff --git a/src/trio/_repl.py b/src/trio/_repl.py
index f9efcc0017..d2a5d818ba 100644
--- a/src/trio/_repl.py
+++ b/src/trio/_repl.py
@@ -12,6 +12,7 @@
 
 import trio
 import trio.lowlevel
+from trio._core._run_context import GLOBAL_RUN_CONTEXT
 from trio._util import final
 
 
@@ -21,10 +22,12 @@ class TrioInteractiveConsole(InteractiveConsole):
     # but when we pass this to FunctionType it expects a dict. So
     # we make the type more specific on our subclass
     locals: dict[str, object]
+    runner: trio._core._run.Runner | None
 
     def __init__(self, repl_locals: dict[str, object] | None = None) -> None:
         super().__init__(locals=repl_locals)
         self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
+        self.runner = None
 
     def runcode(self, code: types.CodeType) -> None:
         func = types.FunctionType(code, self.locals)
@@ -32,6 +35,17 @@ def runcode(self, code: types.CodeType) -> None:
             result = trio.from_thread.run(outcome.acapture, func)
         else:
             result = trio.from_thread.run_sync(outcome.capture, func)
+
+        # clear ki_pending
+        assert self.runner is not None
+        ki_pending = self.runner.ki_pending
+        self.runner.ki_pending = False
+
+        if ki_pending:
+            exc: BaseException | None = KeyboardInterrupt()
+        else:
+            exc = None
+
         if isinstance(result, outcome.Error):
             # If it is SystemExit, quit the repl. Otherwise, print the traceback.
             # If there is a SystemExit inside a BaseExceptionGroup, it probably isn't
@@ -41,21 +55,28 @@ def runcode(self, code: types.CodeType) -> None:
             if isinstance(result.error, SystemExit):
                 raise result.error
             else:
-                # Inline our own version of self.showtraceback that can use
-                # outcome.Error.error directly to print clean tracebacks.
-                # This also means overriding self.showtraceback does nothing.
-                sys.last_type, sys.last_value = type(result.error), result.error
-                sys.last_traceback = result.error.__traceback__
-                # see https://docs.python.org/3/library/sys.html#sys.last_exc
-                if sys.version_info >= (3, 12):
-                    sys.last_exc = result.error
+                if exc:
+                    exc.__context__ = result.error
+                else:
+                    exc = result.error
+
+        if exc:
+            # Inline our own version of self.showtraceback that can use
+            # outcome.Error.error directly to print clean tracebacks.
+            # This also means overriding self.showtraceback does nothing.
+            sys.last_type, sys.last_value = type(exc), exc
+            sys.last_traceback = exc.__traceback__
+            # see https://docs.python.org/3/library/sys.html#sys.last_exc
+            if sys.version_info >= (3, 12):
+                sys.last_exc = exc
 
-                # We always use sys.excepthook, unlike other implementations.
-                # This means that overriding self.write also does nothing to tbs.
-                sys.excepthook(sys.last_type, sys.last_value, sys.last_traceback)
+            # We always use sys.excepthook, unlike other implementations.
+            # This means that overriding self.write also does nothing to tbs.
+            sys.excepthook(sys.last_type, sys.last_value, sys.last_traceback)
 
 
 async def run_repl(console: TrioInteractiveConsole) -> None:
+    console.runner = GLOBAL_RUN_CONTEXT.runner
     banner = (
         f"trio REPL {sys.version} on {sys.platform}\n"
         f'Use "await" directly instead of "trio.run()".\n'

From b6abaf516e79bfe343c78df778b21b31d086508c Mon Sep 17 00:00:00 2001
From: A5rocks <git@helvetica.moe>
Date: Fri, 28 Mar 2025 10:48:51 +0900
Subject: [PATCH 3/7] Polish

---
 src/trio/_core/_io_kqueue.py | 4 ++--
 src/trio/_threads.py         | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py
index dd104e97b4..3eda6b5189 100644
--- a/src/trio/_core/_io_kqueue.py
+++ b/src/trio/_core/_io_kqueue.py
@@ -160,8 +160,8 @@ async def wait_kevent(
             )
         self._registered[key] = _core.current_task()
 
-        def abort(raise_cancel: BaseException) -> Abort:
-            r = abort_func(raise_cancel)
+        def abort(cancel_exc: BaseException) -> Abort:
+            r = abort_func(cancel_exc)
             if r is _core.Abort.SUCCEEDED:  # TODO: test this branch
                 del self._registered[key]
             return r
diff --git a/src/trio/_threads.py b/src/trio/_threads.py
index 4eb97b6495..b00b580d79 100644
--- a/src/trio/_threads.py
+++ b/src/trio/_threads.py
@@ -353,8 +353,8 @@ async def to_thread_run_sync(
     # for the result – or None if this function was cancelled and we should
     # discard the result.
     task_register: list[trio.lowlevel.Task | None] = [trio.lowlevel.current_task()]
-    # Holds a reference to the raise_cancel function provided if a cancellation
-    # is attempted against this task - or None if no such delivery has happened.
+    # Holds a reference to the exception provided if a cancellation is
+    # attempted against this task - or None if no such delivery has happened.
     cancel_register: list[BaseException | None] = [None]
     name = f"trio.to_thread.run_sync-{next(_thread_counter)}"
     placeholder = ThreadPlaceholder(name)

From 6421d8dcac85dd02609af9eb061ee9aeef3cbfab Mon Sep 17 00:00:00 2001
From: A5rocks <git@helvetica.moe>
Date: Fri, 11 Apr 2025 10:17:09 +0900
Subject: [PATCH 4/7] Copy over forgotten deprecation test

---
 src/trio/_core/_tests/test_run.py | 20 ++++++++++++++++++++
 src/trio/_repl.py                 |  2 +-
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/src/trio/_core/_tests/test_run.py b/src/trio/_core/_tests/test_run.py
index 21c6b43a42..86ab585b0f 100644
--- a/src/trio/_core/_tests/test_run.py
+++ b/src/trio/_core/_tests/test_run.py
@@ -1502,6 +1502,26 @@ async def main() -> None:
     assert record == ["main exiting", "2nd ran"]
 
 
+async def test_deprecated_abort_fn_semantics() -> None:
+    with _core.CancelScope() as scope:
+        scope.cancel()
+
+        task = _core.current_task()
+        token = _core.current_trio_token()
+
+        def slow_abort(raise_cancel: Callable[[], NoReturn]) -> _core.Abort:
+            with pytest.warns(
+                DeprecationWarning,
+                match="^wait_task_rescheduled's abort_fn taking a callback argument is deprecated since Trio",
+            ):
+                result = outcome.capture(raise_cancel)
+            token.run_sync_soon(_core.reschedule, task, result)
+            return _core.Abort.FAILED
+
+        with pytest.raises(_core.Cancelled):
+            await _core.wait_task_rescheduled(slow_abort)
+
+
 async def test_slow_abort_basic() -> None:
     with _core.CancelScope() as scope:
         scope.cancel()
diff --git a/src/trio/_repl.py b/src/trio/_repl.py
index 31f97d5361..c5cc305967 100644
--- a/src/trio/_repl.py
+++ b/src/trio/_repl.py
@@ -57,7 +57,7 @@ def runcode(self, code: types.CodeType) -> None:
                 else:
                     exc = result.error
 
-        if exc:
+        if exc is not None:
             # Inline our own version of self.showtraceback that can use
             # outcome.Error.error directly to print clean tracebacks.
             # This also means overriding self.showtraceback does nothing.

From ff3da6bccf85581fb7295a193b140265fda3636d Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Tue, 22 Apr 2025 09:25:37 +0000
Subject: [PATCH 5/7] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 src/trio/_channel.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/trio/_channel.py b/src/trio/_channel.py
index 108dbb2fe4..c10e6946cf 100644
--- a/src/trio/_channel.py
+++ b/src/trio/_channel.py
@@ -17,7 +17,7 @@
 import trio
 
 from ._abc import ReceiveChannel, ReceiveType, SendChannel, SendType, T
-from ._core import Abort, RaiseCancelT, Task, enable_ki_protection
+from ._core import Abort, Task, enable_ki_protection
 from ._util import (
     MultipleExceptionError,
     NoPublicConstructor,

From 261e2b24abe5ef4cd60e8d6ac2f2844185e1de20 Mon Sep 17 00:00:00 2001
From: A5rocks <git@helvetica.moe>
Date: Wed, 7 May 2025 14:49:25 +0900
Subject: [PATCH 6/7] Undo abort function signature change for now

---
 newsfragments/2649.removal.rst          |  4 --
 src/trio/_channel.py                    |  6 +--
 src/trio/_core/_exceptions.py           | 16 +------
 src/trio/_core/_generated_io_kqueue.py  |  4 +-
 src/trio/_core/_io_epoll.py             |  4 +-
 src/trio/_core/_io_kqueue.py            | 10 ++---
 src/trio/_core/_io_windows.py           | 16 +++----
 src/trio/_core/_parking_lot.py          |  2 +-
 src/trio/_core/_run.py                  | 34 ++++++++++----
 src/trio/_core/_tests/test_asyncgen.py  |  2 +-
 src/trio/_core/_tests/test_ki.py        |  9 ++--
 src/trio/_core/_tests/test_run.py       | 34 ++++----------
 src/trio/_core/_traps.py                | 59 ++++++++++++-------------
 src/trio/_subprocess_platform/kqueue.py |  2 +-
 src/trio/_sync.py                       |  3 +-
 src/trio/_tests/test_exports.py         |  8 ----
 src/trio/_threads.py                    | 22 +++++----
 src/trio/_tools/gen_exports.py          |  2 +-
 src/trio/lowlevel.py                    | 15 +------
 19 files changed, 108 insertions(+), 144 deletions(-)
 delete mode 100644 newsfragments/2649.removal.rst

diff --git a/newsfragments/2649.removal.rst b/newsfragments/2649.removal.rst
deleted file mode 100644
index 0c4aee38a7..0000000000
--- a/newsfragments/2649.removal.rst
+++ /dev/null
@@ -1,4 +0,0 @@
-The abort function passed to :func:`~trio.lowlevel.wait_task_rescheduled`
-now directly takes as argument the cancellation exception that should be
-raised after a successful asynchronous cancellation. Previously, it took
-a callable that would raise the exception when called.
diff --git a/src/trio/_channel.py b/src/trio/_channel.py
index c10e6946cf..1ed5945798 100644
--- a/src/trio/_channel.py
+++ b/src/trio/_channel.py
@@ -17,7 +17,7 @@
 import trio
 
 from ._abc import ReceiveChannel, ReceiveType, SendChannel, SendType, T
-from ._core import Abort, Task, enable_ki_protection
+from ._core import Abort, RaiseCancelT, Task, enable_ki_protection
 from ._util import (
     MultipleExceptionError,
     NoPublicConstructor,
@@ -227,7 +227,7 @@ async def send(self, value: SendType) -> None:
         self._state.send_tasks[task] = value
         task.custom_sleep_data = self
 
-        def abort_fn(_: BaseException) -> Abort:
+        def abort_fn(_: RaiseCancelT) -> Abort:
             self._tasks.remove(task)
             del self._state.send_tasks[task]
             return trio.lowlevel.Abort.SUCCEEDED
@@ -375,7 +375,7 @@ async def receive(self) -> ReceiveType:
         self._state.receive_tasks[task] = None
         task.custom_sleep_data = self
 
-        def abort_fn(_: BaseException) -> Abort:
+        def abort_fn(_: RaiseCancelT) -> Abort:
             self._tasks.remove(task)
             del self._state.receive_tasks[task]
             return trio.lowlevel.Abort.SUCCEEDED
diff --git a/src/trio/_core/_exceptions.py b/src/trio/_core/_exceptions.py
index bc01ed577f..4a76a674ac 100644
--- a/src/trio/_core/_exceptions.py
+++ b/src/trio/_core/_exceptions.py
@@ -1,8 +1,7 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, NoReturn
+from typing import TYPE_CHECKING
 
-from trio import _deprecate
 from trio._util import NoPublicConstructor, final
 
 if TYPE_CHECKING:
@@ -71,19 +70,6 @@ class Cancelled(BaseException, metaclass=NoPublicConstructor):
     def __str__(self) -> str:
         return "Cancelled"
 
-    def __call__(self) -> NoReturn:
-        # If a Cancelled exception is passed to an old abort_fn that
-        # expects a raise_cancel callback, someone will eventually try
-        # to call the exception instead of raising it. Provide a
-        # deprecation warning and raise it instead.
-        _deprecate.warn_deprecated(
-            "wait_task_rescheduled's abort_fn taking a callback argument",
-            "0.30.0",
-            issue=2649,
-            instead="an exception argument",
-        )
-        raise self
-
     def __reduce__(self) -> tuple[Callable[[], Cancelled], tuple[()]]:
         return (Cancelled._create, ())
 
diff --git a/src/trio/_core/_generated_io_kqueue.py b/src/trio/_core/_generated_io_kqueue.py
index f942877051..556d29e1f2 100644
--- a/src/trio/_core/_generated_io_kqueue.py
+++ b/src/trio/_core/_generated_io_kqueue.py
@@ -16,7 +16,7 @@
 
     from .. import _core
     from .._file_io import _HasFileNo
-    from ._traps import Abort
+    from ._traps import Abort, RaiseCancelT
 
 assert not TYPE_CHECKING or sys.platform == "darwin"
 
@@ -59,7 +59,7 @@ def monitor_kevent(
 
 @enable_ki_protection
 async def wait_kevent(
-    ident: int, filter: int, abort_func: Callable[[BaseException], Abort]
+    ident: int, filter: int, abort_func: Callable[[RaiseCancelT], Abort]
 ) -> Abort:
     """TODO: these are implemented, but are currently more of a sketch than
     anything real. See `#26
diff --git a/src/trio/_core/_io_epoll.py b/src/trio/_core/_io_epoll.py
index 7287a5be8e..5e05f0813f 100644
--- a/src/trio/_core/_io_epoll.py
+++ b/src/trio/_core/_io_epoll.py
@@ -16,7 +16,7 @@
 if TYPE_CHECKING:
     from typing_extensions import TypeAlias
 
-    from .._core import Abort
+    from .._core import Abort, RaiseCancelT
     from .._file_io import _HasFileNo
 
 
@@ -303,7 +303,7 @@ async def _epoll_wait(self, fd: int | _HasFileNo, attr_name: str) -> None:
         setattr(waiters, attr_name, _core.current_task())
         self._update_registrations(fd)
 
-        def abort(_: BaseException) -> Abort:
+        def abort(_: RaiseCancelT) -> Abort:
             setattr(waiters, attr_name, None)
             self._update_registrations(fd)
             return _core.Abort.SUCCEEDED
diff --git a/src/trio/_core/_io_kqueue.py b/src/trio/_core/_io_kqueue.py
index 3eda6b5189..9718c4df80 100644
--- a/src/trio/_core/_io_kqueue.py
+++ b/src/trio/_core/_io_kqueue.py
@@ -18,7 +18,7 @@
 
     from typing_extensions import TypeAlias
 
-    from .._core import Abort, Task, UnboundedQueue
+    from .._core import Abort, RaiseCancelT, Task, UnboundedQueue
     from .._file_io import _HasFileNo
 
 assert not TYPE_CHECKING or (sys.platform != "linux" and sys.platform != "win32")
@@ -147,7 +147,7 @@ async def wait_kevent(
         self,
         ident: int,
         filter: int,
-        abort_func: Callable[[BaseException], Abort],
+        abort_func: Callable[[RaiseCancelT], Abort],
     ) -> Abort:
         """TODO: these are implemented, but are currently more of a sketch than
         anything real. See `#26
@@ -160,8 +160,8 @@ async def wait_kevent(
             )
         self._registered[key] = _core.current_task()
 
-        def abort(cancel_exc: BaseException) -> Abort:
-            r = abort_func(cancel_exc)
+        def abort(raise_cancel: RaiseCancelT) -> Abort:
+            r = abort_func(raise_cancel)
             if r is _core.Abort.SUCCEEDED:  # TODO: test this branch
                 del self._registered[key]
             return r
@@ -180,7 +180,7 @@ async def _wait_common(
         event = select.kevent(fd, filter, flags)
         self._kqueue.control([event], 0)
 
-        def abort(_: BaseException) -> Abort:
+        def abort(_: RaiseCancelT) -> Abort:
             event = select.kevent(fd, filter, select.KQ_EV_DELETE)
             try:
                 self._kqueue.control([event], 0)
diff --git a/src/trio/_core/_io_windows.py b/src/trio/_core/_io_windows.py
index cbfae9bfe2..148253ab88 100644
--- a/src/trio/_core/_io_windows.py
+++ b/src/trio/_core/_io_windows.py
@@ -45,7 +45,7 @@
     from typing_extensions import Buffer, TypeAlias
 
     from .._file_io import _HasFileNo
-    from ._traps import Abort
+    from ._traps import Abort, RaiseCancelT
     from ._unbounded_queue import UnboundedQueue
 
 EventResult: TypeAlias = int
@@ -752,7 +752,7 @@ async def _afd_poll(self, sock: _HasFileNo | int, mode: str) -> None:
         # we let it escape.
         self._refresh_afd(base_handle)
 
-        def abort_fn(_: BaseException) -> Abort:
+        def abort_fn(_: RaiseCancelT) -> Abort:
             setattr(waiters, mode, None)
             self._refresh_afd(base_handle)
             return _core.Abort.SUCCEEDED
@@ -864,11 +864,11 @@ async def wait_overlapped(
             )
         task = _core.current_task()
         self._overlapped_waiters[lpOverlapped] = task
-        cancel_exc = None
+        raise_cancel = None
 
-        def abort(cancel_exc_: BaseException) -> Abort:
-            nonlocal cancel_exc
-            cancel_exc = cancel_exc_
+        def abort(raise_cancel_: RaiseCancelT) -> Abort:
+            nonlocal raise_cancel
+            raise_cancel = raise_cancel_
             try:
                 _check(kernel32.CancelIoEx(handle, lpOverlapped))
             except OSError as exc:
@@ -914,8 +914,8 @@ def abort(cancel_exc_: BaseException) -> Abort:
             # it will produce the right sorts of exceptions
             code = ntdll.RtlNtStatusToDosError(lpOverlappedTyped.Internal)
             if code == ErrorCodes.ERROR_OPERATION_ABORTED:
-                if cancel_exc is not None:
-                    raise cancel_exc
+                if raise_cancel is not None:
+                    raise_cancel()
                 else:
                     # We didn't request this cancellation, so assume
                     # it happened due to the underlying handle being
diff --git a/src/trio/_core/_parking_lot.py b/src/trio/_core/_parking_lot.py
index efd6e83fc0..ddf6276117 100644
--- a/src/trio/_core/_parking_lot.py
+++ b/src/trio/_core/_parking_lot.py
@@ -183,7 +183,7 @@ async def park(self) -> None:
         self._parked[task] = None
         task.custom_sleep_data = self
 
-        def abort_fn(_: BaseException) -> _core.Abort:
+        def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
             del task.custom_sleep_data._parked[task]
             return _core.Abort.SUCCEEDED
 
diff --git a/src/trio/_core/_run.py b/src/trio/_core/_run.py
index 83ab02ee1e..680ea409a2 100644
--- a/src/trio/_core/_run.py
+++ b/src/trio/_core/_run.py
@@ -1228,10 +1228,12 @@ async def _nested_child_finished(
             # If we have a KeyboardInterrupt injected, we want to save it in
             # the nursery's final exceptions list. But if it's just a
             # Cancelled, then we don't -- see gh-1457.
-            def aborted(exn: BaseException) -> Abort:
-                # TODO: when do we abort with something other than Cancelled?
+            def aborted(raise_cancel: _core.RaiseCancelT) -> Abort:
+                exn = capture(raise_cancel).error
                 if not isinstance(exn, Cancelled):
                     self._add_exc(exn)
+                # see test_cancel_scope_exit_doesnt_create_cyclic_garbage
+                del exn  # prevent cyclic garbage creation
                 return Abort.FAILED
 
             self._parent_waiting_in_aexit = True
@@ -1434,7 +1436,7 @@ class Task(metaclass=NoPublicConstructor):  # type: ignore[explicit-any]
     # Tasks start out unscheduled.
     _next_send_fn: Callable[[Any], object] | None = None  # type: ignore[explicit-any]
     _next_send: Outcome[Any] | BaseException | None = None  # type: ignore[explicit-any]
-    _abort_func: Callable[[BaseException], Abort] | None = None
+    _abort_func: Callable[[_core.RaiseCancelT], Abort] | None = None
     custom_sleep_data: Any = None  # type: ignore[explicit-any]
 
     # For introspection and nursery.start()
@@ -1548,24 +1550,24 @@ def _activate_cancel_status(self, cancel_status: CancelStatus | None) -> None:
             if self._cancel_status.effectively_cancelled:
                 self._attempt_delivery_of_any_pending_cancel()
 
-    def _attempt_abort(self, cancel_exc: BaseException) -> None:
+    def _attempt_abort(self, raise_cancel: _core.RaiseCancelT) -> None:
         # Either the abort succeeds, in which case we will reschedule the
         # task, or else it fails, in which case it will worry about
-        # rescheduling itself (hopefully eventually calling raising
+        # rescheduling itself (hopefully eventually calling reraise to raise
         # the given exception, but not necessarily).
 
         # This is only called by the functions immediately below, which both check
         # `self.abort_func is not None`.
         assert self._abort_func is not None, "FATAL INTERNAL ERROR"
 
-        success = self._abort_func(cancel_exc)
+        success = self._abort_func(raise_cancel)
         if type(success) is not Abort:
             raise TrioInternalError("abort function must return Abort enum")
         # We only attempt to abort once per blocking call, regardless of
         # whether we succeeded or failed.
         self._abort_func = None
         if success is Abort.SUCCEEDED:
-            self._runner.reschedule(self, Error(cancel_exc))
+            self._runner.reschedule(self, capture(raise_cancel))
 
     def _attempt_delivery_of_any_pending_cancel(self) -> None:
         if self._abort_func is None:
@@ -1573,7 +1575,21 @@ def _attempt_delivery_of_any_pending_cancel(self) -> None:
         if not self._cancel_status.effectively_cancelled:
             return
 
-        self._attempt_abort(Cancelled._create())
+        def raise_cancel() -> NoReturn:
+            raise Cancelled._create()
+
+        self._attempt_abort(raise_cancel)
+
+    def _attempt_delivery_of_pending_ki(self) -> None:
+        assert self._runner.ki_pending
+        if self._abort_func is None:
+            return
+
+        def raise_cancel() -> NoReturn:
+            self._runner.ki_pending = False
+            raise KeyboardInterrupt
+
+        self._attempt_abort(raise_cancel)
 
 
 ################################################################
@@ -2169,7 +2185,7 @@ async def test_lock_fairness():
         key = (cushion, id(task))
         self.waiting_for_idle[key] = task
 
-        def abort(_: BaseException) -> Abort:
+        def abort(_: _core.RaiseCancelT) -> Abort:
             del self.waiting_for_idle[key]
             return Abort.SUCCEEDED
 
diff --git a/src/trio/_core/_tests/test_asyncgen.py b/src/trio/_core/_tests/test_asyncgen.py
index 29b0aa5e5a..8147a0e57b 100644
--- a/src/trio/_core/_tests/test_asyncgen.py
+++ b/src/trio/_core/_tests/test_asyncgen.py
@@ -255,7 +255,7 @@ async def step_outside_async_context(aiter_: AsyncGenerator[int, None]) -> None:
     # NB: the strangeness with aiter being an attribute of abort_fn is
     # to make it as easy as possible to ensure we don't hang onto a
     # reference to aiter inside the guts of the run loop.
-    def abort_fn(_: BaseException) -> _core.Abort:
+    def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
         with pytest.raises(StopIteration, match="42"):
             abort_fn.aiter.asend(None).send(None)  # type: ignore[attr-defined]  # Callables don't have attribute "aiter"
         del abort_fn.aiter  # type: ignore[attr-defined]
diff --git a/src/trio/_core/_tests/test_ki.py b/src/trio/_core/_tests/test_ki.py
index 4b7e329879..40f2a7f40b 100644
--- a/src/trio/_core/_tests/test_ki.py
+++ b/src/trio/_core/_tests/test_ki.py
@@ -36,7 +36,7 @@
         Iterator,
     )
 
-    from ..._core import Abort
+    from ..._core import Abort, RaiseCancelT
 
 
 def ki_self() -> None:
@@ -407,7 +407,7 @@ async def main_4() -> None:
         ki_self()
         task = _core.current_task()
 
-        def abort(_: BaseException) -> Abort:
+        def abort(_: RaiseCancelT) -> Abort:
             _core.reschedule(task, outcome.Value(1))
             return _core.Abort.FAILED
 
@@ -427,8 +427,9 @@ async def main_5() -> None:
         ki_self()
         task = _core.current_task()
 
-        def abort(cancel_exc: BaseException) -> Abort:
-            _core.reschedule(task, outcome.Error(cancel_exc))
+        def abort(raise_cancel: RaiseCancelT) -> Abort:
+            result = outcome.capture(raise_cancel)
+            _core.reschedule(task, result)
             return _core.Abort.FAILED
 
         with pytest.raises(_core.Cancelled):
diff --git a/src/trio/_core/_tests/test_run.py b/src/trio/_core/_tests/test_run.py
index fabb1c66f9..c02d185d45 100644
--- a/src/trio/_core/_tests/test_run.py
+++ b/src/trio/_core/_tests/test_run.py
@@ -1503,26 +1503,6 @@ async def main() -> None:
     assert record == ["main exiting", "2nd ran"]
 
 
-async def test_deprecated_abort_fn_semantics() -> None:
-    with _core.CancelScope() as scope:
-        scope.cancel()
-
-        task = _core.current_task()
-        token = _core.current_trio_token()
-
-        def slow_abort(raise_cancel: Callable[[], NoReturn]) -> _core.Abort:
-            with pytest.warns(
-                DeprecationWarning,
-                match="^wait_task_rescheduled's abort_fn taking a callback argument is deprecated since Trio",
-            ):
-                result = outcome.capture(raise_cancel)
-            token.run_sync_soon(_core.reschedule, task, result)
-            return _core.Abort.FAILED
-
-        with pytest.raises(_core.Cancelled):
-            await _core.wait_task_rescheduled(slow_abort)
-
-
 async def test_slow_abort_basic() -> None:
     with _core.CancelScope() as scope:
         scope.cancel()
@@ -1530,8 +1510,9 @@ async def test_slow_abort_basic() -> None:
         task = _core.current_task()
         token = _core.current_trio_token()
 
-        def slow_abort(cancel_exc: BaseException) -> _core.Abort:
-            token.run_sync_soon(_core.reschedule, task, outcome.Error(cancel_exc))
+        def slow_abort(raise_cancel: _core.RaiseCancelT) -> _core.Abort:
+            result = outcome.capture(raise_cancel)
+            token.run_sync_soon(_core.reschedule, task, result)
             return _core.Abort.FAILED
 
         with pytest.raises(_core.Cancelled):
@@ -1545,9 +1526,10 @@ async def slow_aborter() -> None:
         task = _core.current_task()
         token = _core.current_trio_token()
 
-        def slow_abort(cancel_exc: BaseException) -> _core.Abort:
+        def slow_abort(raise_cancel: _core.RaiseCancelT) -> _core.Abort:
             record.append("abort-called")
-            token.run_sync_soon(_core.reschedule, task, outcome.Error(cancel_exc))
+            result = outcome.capture(raise_cancel)
+            token.run_sync_soon(_core.reschedule, task, result)
             return _core.Abort.FAILED
 
         record.append("sleeping")
@@ -2386,7 +2368,7 @@ async def reattachable_coroutine() -> None:
 
         task = _core.current_task()
 
-        def abort_fn(_: BaseException) -> _core.Abort:  # pragma: no cover
+        def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:  # pragma: no cover
             return _core.Abort.FAILED
 
         got = await _core.temporarily_detach_coroutine_object(abort_fn)
@@ -2430,7 +2412,7 @@ async def reattachable_coroutine() -> None:
         nonlocal task
         task = _core.current_task()
 
-        def abort_fn(_: BaseException) -> _core.Abort:
+        def abort_fn(_: _core.RaiseCancelT) -> _core.Abort:
             nonlocal abort_fn_called
             abort_fn_called = True
             return _core.Abort.FAILED
diff --git a/src/trio/_core/_traps.py b/src/trio/_core/_traps.py
index ba5b345bfe..60f72d1295 100644
--- a/src/trio/_core/_traps.py
+++ b/src/trio/_core/_traps.py
@@ -32,7 +32,7 @@ class CancelShieldedCheckpoint:
 # Not exported in the trio._core namespace, but imported directly by _run.
 @attrs.frozen(slots=False)
 class WaitTaskRescheduled:
-    abort_func: Callable[[BaseException], Abort]
+    abort_func: Callable[[RaiseCancelT], Abort]
 
 
 # Not exported in the trio._core namespace, but imported directly by _run.
@@ -105,7 +105,7 @@ class Abort(enum.Enum):
 # Should always return the type a Task "expects", unless you willfully reschedule it
 # with a bad value.
 async def wait_task_rescheduled(  # type: ignore[explicit-any]
-    abort_func: Callable[[BaseException], Abort],
+    abort_func: Callable[[RaiseCancelT], Abort],
 ) -> Any:
     """Put the current task to sleep, with cancellation support.
 
@@ -137,7 +137,7 @@ async def wait_task_rescheduled(  # type: ignore[explicit-any]
        timeout expiring). When this happens, the ``abort_func`` is called. Its
        interface looks like::
 
-           def abort_func(exc):
+           def abort_func(raise_cancel):
                ...
                return trio.lowlevel.Abort.SUCCEEDED  # or FAILED
 
@@ -151,43 +151,40 @@ def abort_func(exc):
        task can't be cancelled at this time, and still has to make sure that
        "someone" eventually calls :func:`reschedule`.
 
-       At that point there are again two possibilities. You can simply
-       ignore the cancellation altogether: wait for the operation to
-       complete and then reschedule and continue as normal. (For
-       example, this is what :func:`trio.to_thread.run_sync` does if
-       cancellation is disabled.)  The other possibility is that the
-       ``abort_func`` does succeed in cancelling the operation, but
-       for some reason isn't able to report that right away. (Example:
-       on Windows, it's possible to request that an async
-       ("overlapped") I/O operation be cancelled, but this request is
-       *also* asynchronous – you don't find out until later whether
-       the operation was actually cancelled or not.)  To report a
-       delayed cancellation, you should reschedule the task yourself,
-       and cause it to raise the exception ``exc`` that was passed to
-       ``abort_func``. (Currently ``exc`` will always be a
-       `~trio.Cancelled` exception, but we may use this mechanism to
-       raise other exceptions in the future; you should raise whatever
-       you're given.) Either of the approaches sketched below can
-       work::
+       At that point there are again two possibilities. You can simply ignore
+       the cancellation altogether: wait for the operation to complete and
+       then reschedule and continue as normal. (For example, this is what
+       :func:`trio.to_thread.run_sync` does if cancellation is disabled.)
+       The other possibility is that the ``abort_func`` does succeed in
+       cancelling the operation, but for some reason isn't able to report that
+       right away. (Example: on Windows, it's possible to request that an
+       async ("overlapped") I/O operation be cancelled, but this request is
+       *also* asynchronous – you don't find out until later whether the
+       operation was actually cancelled or not.)  To report a delayed
+       cancellation, then you should reschedule the task yourself, and call
+       the ``raise_cancel`` callback passed to ``abort_func`` to raise a
+       :exc:`~trio.Cancelled` (or possibly :exc:`KeyboardInterrupt`) exception
+       into this task. Either of the approaches sketched below can work::
 
           # Option 1:
-          # Directly reschedule the task with the provided exception.
+          # Catch the exception from raise_cancel and inject it into the task.
           # (This is what Trio does automatically for you if you return
           # Abort.SUCCEEDED.)
-          trio.lowlevel.reschedule(task, outcome.Error(exc))
+          trio.lowlevel.reschedule(task, outcome.capture(raise_cancel))
 
           # Option 2:
           # wait to be woken by "someone", and then decide whether to raise
           # the error from inside the task.
-          outer_exc = None
-          def abort(inner_exc):
-              nonlocal outer_exc
-              outer_exc = inner_exc
+          outer_raise_cancel = None
+          def abort(inner_raise_cancel):
+              nonlocal outer_raise_cancel
+              outer_raise_cancel = inner_raise_cancel
               TRY_TO_CANCEL_OPERATION()
               return trio.lowlevel.Abort.FAILED
           await wait_task_rescheduled(abort)
           if OPERATION_WAS_SUCCESSFULLY_CANCELLED:
-              raise outer_exc
+              # raises the error
+              outer_raise_cancel()
 
        In any case it's guaranteed that we only call the ``abort_func`` at most
        once per call to :func:`wait_task_rescheduled`.
@@ -245,7 +242,7 @@ async def permanently_detach_coroutine_object(
 
 
 async def temporarily_detach_coroutine_object(
-    abort_func: Callable[[BaseException], Abort],
+    abort_func: Callable[[RaiseCancelT], Abort],
 ) -> object:
     """Temporarily detach the current coroutine object from the Trio
     scheduler.
@@ -272,8 +269,8 @@ async def temporarily_detach_coroutine_object(
           detached task directly without going through
           :func:`reattach_detached_coroutine_object`, which would be bad.)
           Your ``abort_func`` should still arrange for whatever the coroutine
-          object is doing to be cancelled, and then reattach to Trio and raise
-          the exception it received, if possible.
+          object is doing to be cancelled, and then reattach to Trio and call
+          the ``raise_cancel`` callback, if possible.
 
     Returns or raises whatever value or exception the new coroutine runner
     uses to resume the coroutine.
diff --git a/src/trio/_subprocess_platform/kqueue.py b/src/trio/_subprocess_platform/kqueue.py
index 2b257eeea5..2283bb5360 100644
--- a/src/trio/_subprocess_platform/kqueue.py
+++ b/src/trio/_subprocess_platform/kqueue.py
@@ -41,7 +41,7 @@ def make_event(flags: int) -> select.kevent:
         # in Chromium it seems we should still keep the check.
         return
 
-    def abort(_: BaseException) -> _core.Abort:
+    def abort(_: _core.RaiseCancelT) -> _core.Abort:
         kqueue.control([make_event(select.KQ_EV_DELETE)], 0)
         return _core.Abort.SUCCEEDED
 
diff --git a/src/trio/_sync.py b/src/trio/_sync.py
index 383dd4168a..ca373922b0 100644
--- a/src/trio/_sync.py
+++ b/src/trio/_sync.py
@@ -11,6 +11,7 @@
 from ._core import (
     Abort,
     ParkingLot,
+    RaiseCancelT,
     add_parking_lot_breaker,
     enable_ki_protection,
     remove_parking_lot_breaker,
@@ -94,7 +95,7 @@ async def wait(self) -> None:
             task = _core.current_task()
             self._tasks.add(task)
 
-            def abort_fn(_: BaseException) -> Abort:
+            def abort_fn(_: RaiseCancelT) -> Abort:
                 self._tasks.remove(task)
                 return _core.Abort.SUCCEEDED
 
diff --git a/src/trio/_tests/test_exports.py b/src/trio/_tests/test_exports.py
index d69fad1104..0bf619e3b8 100644
--- a/src/trio/_tests/test_exports.py
+++ b/src/trio/_tests/test_exports.py
@@ -65,7 +65,6 @@ def test_core_is_properly_reexported() -> None:
     # Each export from _core should be re-exported by exactly one of these
     # three modules:
     sources = [trio, trio.lowlevel, trio.testing]
-    hit_RaiseCancelT = False
     for symbol in dir(_core):
         if symbol.startswith("_"):
             continue
@@ -77,15 +76,8 @@ def test_core_is_properly_reexported() -> None:
             ):
                 found += 1
         print(symbol, found)
-        if symbol == "RaiseCancelT":
-            # deprecated
-            hit_RaiseCancelT = True
-            continue
-
         assert found == 1
 
-    assert hit_RaiseCancelT
-
 
 def class_is_final(cls: type) -> bool:
     """Check if a class cannot be subclassed."""
diff --git a/src/trio/_threads.py b/src/trio/_threads.py
index b00b580d79..f985c69309 100644
--- a/src/trio/_threads.py
+++ b/src/trio/_threads.py
@@ -31,6 +31,8 @@
 
     from typing_extensions import TypeVarTuple, Unpack
 
+    from trio._core._traps import RaiseCancelT
+
     Ts = TypeVarTuple("Ts")
 
 RetT = TypeVar("RetT")
@@ -42,7 +44,7 @@ class _ParentTaskData(threading.local):
 
     token: TrioToken
     abandon_on_cancel: bool
-    cancel_register: list[BaseException | None]
+    cancel_register: list[RaiseCancelT | None]
     task_register: list[trio.lowlevel.Task | None]
 
 
@@ -349,13 +351,15 @@ async def to_thread_run_sync(
     if limiter is None:
         limiter = current_default_thread_limiter()
 
+    # TODO: cancel_register can probably be a single element tuple for typing reasons
+
     # Holds a reference to the task that's blocked in this function waiting
     # for the result – or None if this function was cancelled and we should
     # discard the result.
     task_register: list[trio.lowlevel.Task | None] = [trio.lowlevel.current_task()]
-    # Holds a reference to the exception provided if a cancellation is
-    # attempted against this task - or None if no such delivery has happened.
-    cancel_register: list[BaseException | None] = [None]
+    # Holds a reference to the raise_cancel function provided if a cancellation
+    # is attempted against this task - or None if no such delivery has happened.
+    cancel_register: list[RaiseCancelT | None] = [None]  # type: ignore[assignment]
     name = f"trio.to_thread.run_sync-{next(_thread_counter)}"
     placeholder = ThreadPlaceholder(name)
 
@@ -426,9 +430,9 @@ def deliver_worker_fn_result(result: outcome.Outcome[RetT]) -> None:
             limiter.release_on_behalf_of(placeholder)
             raise
 
-        def abort(cancel_exc: BaseException) -> trio.lowlevel.Abort:
+        def abort(raise_cancel: RaiseCancelT) -> trio.lowlevel.Abort:
             # fill so from_thread_check_cancelled can raise
-            cancel_register[0] = cancel_exc
+            cancel_register[0] = raise_cancel
             if abandon_bool:
                 # empty so report_back_in_trio_thread_fn cannot reschedule
                 task_register[0] = None
@@ -482,13 +486,13 @@ def from_thread_check_cancelled() -> None:
        for completeness.
     """
     try:
-        cancel_exc = PARENT_TASK_DATA.cancel_register[0]
+        raise_cancel = PARENT_TASK_DATA.cancel_register[0]
     except AttributeError:
         raise RuntimeError(
             "this thread wasn't created by Trio, can't check for cancellation",
         ) from None
-    if cancel_exc is not None:
-        raise cancel_exc
+    if raise_cancel is not None:
+        raise_cancel()
 
 
 def _send_message_to_trio(
diff --git a/src/trio/_tools/gen_exports.py b/src/trio/_tools/gen_exports.py
index f603a7f3ff..5b1affe24a 100755
--- a/src/trio/_tools/gen_exports.py
+++ b/src/trio/_tools/gen_exports.py
@@ -382,7 +382,7 @@ def main() -> None:  # pragma: no cover
 
     from .. import _core
     from .._file_io import _HasFileNo
-    from ._traps import Abort
+    from ._traps import Abort, RaiseCancelT
 """
 
 IMPORTS_WINDOWS = """\
diff --git a/src/trio/lowlevel.py b/src/trio/lowlevel.py
index 7aece2fb4f..bbeab6af17 100644
--- a/src/trio/lowlevel.py
+++ b/src/trio/lowlevel.py
@@ -16,7 +16,7 @@
     Abort as Abort,
     ParkingLot as ParkingLot,
     ParkingLotStatistics as ParkingLotStatistics,
-    RaiseCancelT as _RaiseCancelT,
+    RaiseCancelT as RaiseCancelT,
     RunStatistics as RunStatistics,
     RunVar as RunVar,
     RunVarToken as RunVarToken,
@@ -84,15 +84,4 @@
             wait_kevent as wait_kevent,
         )
 
-from ._deprecate import DeprecatedAttribute, deprecate_attributes
-
-deprecate_attributes(
-    __name__,
-    {
-        "RaiseCancelT": DeprecatedAttribute(
-            _RaiseCancelT, "0.30.0", issue=2649, instead="an exception argument"
-        )
-    },
-)
-
-del sys, deprecate_attributes, DeprecatedAttribute
+del sys

From f83ebbb5364f479187cdf64066e110f377255f63 Mon Sep 17 00:00:00 2001
From: A5rocks <git@helvetica.moe>
Date: Wed, 11 Jun 2025 12:34:33 +0900
Subject: [PATCH 7/7] Fix faulty merge and update TODOs

`pytest-trio` should error if `ki_pending` is `True` at the end of
execution...
---
 src/trio/_core/_run.py                  | 11 -----------
 src/trio/_core/_tests/test_cancelled.py | 12 ++++++------
 src/trio/_core/_tests/test_ki.py        |  3 ++-
 src/trio/_threads.py                    |  2 --
 4 files changed, 8 insertions(+), 20 deletions(-)

diff --git a/src/trio/_core/_run.py b/src/trio/_core/_run.py
index dc6d09bdcc..dc82fb275a 100644
--- a/src/trio/_core/_run.py
+++ b/src/trio/_core/_run.py
@@ -1662,17 +1662,6 @@ def raise_cancel() -> NoReturn:
 
         self._attempt_abort(raise_cancel)
 
-    def _attempt_delivery_of_pending_ki(self) -> None:
-        assert self._runner.ki_pending
-        if self._abort_func is None:
-            return
-
-        def raise_cancel() -> NoReturn:
-            self._runner.ki_pending = False
-            raise KeyboardInterrupt
-
-        self._attempt_abort(raise_cancel)
-
 
 ################################################################
 # The central Runner object
diff --git a/src/trio/_core/_tests/test_cancelled.py b/src/trio/_core/_tests/test_cancelled.py
index 0c144c37f0..1be9f2c842 100644
--- a/src/trio/_core/_tests/test_cancelled.py
+++ b/src/trio/_core/_tests/test_cancelled.py
@@ -199,14 +199,12 @@ async def child() -> None:
             raise ValueError
 
 
-async def test_reason_delayed_ki() -> None:
+def test_reason_delayed_ki() -> None:
     # simplified version of test_ki.test_ki_protection_works check #2
-    parent_task = current_task()
-
     async def sleeper(name: str) -> None:
         with pytest.raises(
             Cancelled,
-            match=rf"^cancelled due to KeyboardInterrupt from task {parent_task!r}$",
+            match=r"^cancelled due to KeyboardInterrupt$",
         ):
             while True:
                 await trio.lowlevel.checkpoint()
@@ -214,9 +212,11 @@ async def sleeper(name: str) -> None:
     async def raiser(name: str) -> None:
         ki_self()
 
-    with RaisesGroup(KeyboardInterrupt):
+    async def main() -> None:
         async with trio.open_nursery() as nursery:
             nursery.start_soon(sleeper, "s1")
             nursery.start_soon(sleeper, "s2")
             nursery.start_soon(trio.lowlevel.enable_ki_protection(raiser), "r1")
-            # __aexit__ blocks, and then receives the KI
+
+    with pytest.raises(KeyboardInterrupt):
+        trio.run(main)
diff --git a/src/trio/_core/_tests/test_ki.py b/src/trio/_core/_tests/test_ki.py
index 40f2a7f40b..5ff44ec804 100644
--- a/src/trio/_core/_tests/test_ki.py
+++ b/src/trio/_core/_tests/test_ki.py
@@ -329,7 +329,8 @@ async def check_protected_kill() -> None:
     with pytest.raises(KeyboardInterrupt) as excinfo:
         _core.run(check_protected_kill)
 
-    # TODO: be consistent about providing Cancelled tree as __context__
+    # TODO: consider ensuring `__context__` is `None` in all cases above
+    # and below if the tree of `Cancelled`s is very spammy.
     assert excinfo.value.__context__ is None
     assert record_set == {"s1 ok", "s2 ok", "r1 cancel ok"}
 
diff --git a/src/trio/_threads.py b/src/trio/_threads.py
index 930226d8a5..4b1e54f540 100644
--- a/src/trio/_threads.py
+++ b/src/trio/_threads.py
@@ -351,8 +351,6 @@ async def to_thread_run_sync(
     if limiter is None:
         limiter = current_default_thread_limiter()
 
-    # TODO: cancel_register can probably be a single element tuple for typing reasons
-
     # Holds a reference to the task that's blocked in this function waiting
     # for the result – or None if this function was cancelled and we should
     # discard the result.