@@ -1,17 +1,19 @@
-"""Cancellation utilities for cleaning up background tasks.
+"""Cancellation-correct waiting primitives.
 
 When we cancel one of our own background tasks (a heartbeat, a monitor, a
 renewal loop) and await it, the task's CancelledError surfaces in the awaiter
 and must not propagate: we asked for it.  A CancelledError should only
 propagate when it was delivered to the *caller* -- an external cancellation
-that arrived while we were waiting.
-
-Telling the two apart by the cancel message (task.cancel(msg=...)) is
-unreliable: CPython drops the message when the task's awaited future completes
-in the same event-loop tick as the cancellation (python/cpython#91048), and on
-Python 3.10 the message never propagates at all.  The awaited task's final
-state is reliable on every supported version, so cancel_task() checks that
-instead.
+that arrived while we were waiting.  The helpers here guarantee both halves
+of that contract, which plain awaits and asyncio.wait_for do not.
+
+Awaiting the task directly cannot tell those two apart: the cancel message
+(task.cancel(msg=...)) is dropped when the awaited future completes in the
+same event-loop tick as the cancellation (python/cpython#91048), and the
+task's final state is ambiguous when both cancellations happen at once.
+asyncio.wait() never raises the awaited task's CancelledError at all, so the
+only CancelledError that can escape cancel_task() is one delivered to the
+caller, which must propagate.
 """
 
 import asyncio
@@ -21,21 +23,39 @@
 CANCEL_MSG_CLEANUP = "docket:cleanup"
 
 
+async def _wait_for_event(event: asyncio.Event, timeout: float) -> bool:
+    """Wait up to ``timeout`` seconds for ``event``; True if it was set.
+
+    asyncio.wait_for on Python 3.10 and 3.11 swallows a cancellation that is
+    delivered to the caller in the same event-loop tick that the inner future
+    completes (python/cpython#86296; the 3.12 rewrite fixed it).  The worker
+    waits on events between polls, so a worker cancelled at that moment would
+    lose the cancellation and run forever.  asyncio.wait never returns the
+    inner task's result through the caller and never absorbs the caller's
+    cancellation.
+    """
+    if event.is_set():
+        return True
+    waiter = asyncio.ensure_future(event.wait())
+    try:
+        done, _ = await asyncio.wait([waiter], timeout=timeout)
+        return bool(done)
+    finally:
+        waiter.cancel()
+
+
 async def cancel_task(task: "asyncio.Task[Any]", reason: str) -> None:
     """Cancel a task and await its completion, suppressing its cancellation.
 
-    If the awaited task ended cancelled, the CancelledError came from it, and
-    we swallow it -- we initiated that cancellation.  If the task did not end
-    cancelled, the CancelledError was delivered to the caller instead, so it
-    propagates.
+    A CancelledError raised here was delivered to the caller, not the awaited
+    task, and propagates.  If the task ended with a real exception instead of
+    cancelling, that exception is re-raised.
 
     Args:
         task: The task to cancel
         reason: A description of why we're cancelling (e.g., CANCEL_MSG_CLEANUP)
     """
     task.cancel(reason)
-    try:
-        await task
-    except asyncio.CancelledError:
-        if not task.cancelled():  # pragma: no cover - caller cancelled mid-await
-            raise
+    await asyncio.wait([task])
+    if not task.cancelled() and (error := task.exception()) is not None:
+        raise error
@@ -34,7 +34,7 @@
 from opentelemetry import trace
 from opentelemetry.trace import Status, StatusCode, Tracer
 
-from ._cancellation import CANCEL_MSG_CLEANUP, cancel_task
+from ._cancellation import CANCEL_MSG_CLEANUP, _wait_for_event, cancel_task
 from ._lua import Arg, Key, redis_script
 from ._redis import RedisClient
 from ._telemetry import suppress_instrumentation
@@ -109,14 +109,6 @@
 TaskKey: TypeAlias = str
 
 
-async def _wait_for_event(event: asyncio.Event, timeout: float) -> bool:
-    try:
-        await asyncio.wait_for(event.wait(), timeout=timeout)
-    except asyncio.TimeoutError:
-        return False
-    return True
-
-
 class PubSubMessage(TypedDict):
     """Message received from Redis pub/sub pattern subscription."""
 
 
@@ -0,0 +1,139 @@
+"""Regression tests: cancelling a worker externally must always shut it down.
+
+asyncio.wait_for on Python 3.10 and 3.11 swallows a cancellation that is
+delivered to the caller in the same event-loop tick that the inner future
+completes (python/cpython#86296; the 3.12 rewrite fixed it).  The worker
+waits on events between polls, so a worker cancelled at exactly that moment
+lost the cancellation and ran forever.  These tests cancel at a range of
+event-loop tick offsets so at least one case lands inside that window.
+"""
+
+import asyncio
+
+import pytest
+
+from docket import Docket, Worker
+from docket._cancellation import (
+    _wait_for_event,  # pyright: ignore[reportPrivateUsage]
+    cancel_task,
+)
+
+OFFSETS = [0, 1, 2, 3]
+
+
+async def test_wait_for_event_returns_true_for_a_set_event():
+    """An already-set event returns True without waiting"""
+    event = asyncio.Event()
+    event.set()
+
+    assert await _wait_for_event(event, 10.0) is True
+
+
+async def test_wait_for_event_returns_false_on_timeout():
+    """An event that never sets returns False after the timeout"""
+    event = asyncio.Event()
+
+    assert await _wait_for_event(event, 0.01) is False
+
+
+async def test_cancel_task_reraises_a_real_exception():
+    """A subtask that fails instead of cancelling re-raises its exception"""
+
+    async def defiant() -> None:
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            raise RuntimeError("cleanup went sideways") from None
+
+    task = asyncio.create_task(defiant())
+    await asyncio.sleep(0)
+
+    with pytest.raises(RuntimeError, match="cleanup went sideways"):
+        await cancel_task(task, "test cleanup")
+
+
+async def test_cancel_task_tolerates_a_task_that_finishes_anyway():
+    """A subtask that absorbs the cancellation and returns is left alone"""
+
+    async def stubborn() -> None:
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            pass
+
+    task = asyncio.create_task(stubborn())
+    await asyncio.sleep(0)
+
+    await cancel_task(task, "test cleanup")
+
+    assert task.done()
+    assert not task.cancelled()
+
+
+async def _wait_for_worker_readiness(worker: Worker) -> None:
+    """Yield until the worker's cancellation listener reports readiness."""
+    deadline = asyncio.get_running_loop().time() + 5
+    while asyncio.get_running_loop().time() < deadline:
+        session = worker._processing_session  # pyright: ignore[reportPrivateUsage]
+        if session is not None and session.cancellation_ready.is_set():
+            return
+        await asyncio.sleep(0)
+    raise TimeoutError("worker never became ready")
+
+
+@pytest.mark.parametrize("offset", OFFSETS)
+async def test_wait_for_event_honors_caller_cancellation(offset: int):
+    """A cancellation that races the event being set still cancels the caller"""
+    event = asyncio.Event()
+    task = asyncio.create_task(_wait_for_event(event, 10.0))
+    await asyncio.sleep(0)
+
+    event.set()
+    for _ in range(offset):
+        await asyncio.sleep(0)
+    cancel_requested = task.cancel()
+
+    await asyncio.wait([task], timeout=5)
+
+    assert task.done(), "the wait outlived its cancellation"
+    assert not cancel_requested or task.cancelled(), (
+        "the caller's cancellation was swallowed"
+    )
+
+
+@pytest.mark.parametrize("offset", OFFSETS)
+async def test_cancel_task_propagates_caller_cancellation(offset: int):
+    """A cancellation that races the subtask's own cancellation still propagates"""
+    forever = asyncio.Event()
+    inner = asyncio.create_task(forever.wait())
+    await asyncio.sleep(0)
+
+    outer = asyncio.create_task(cancel_task(inner, "test cleanup"))
+    for _ in range(offset):
+        await asyncio.sleep(0)
+    cancel_requested = outer.cancel()
+
+    await asyncio.wait([outer], timeout=5)
+
+    assert outer.done(), "cancel_task outlived its cancellation"
+    assert not cancel_requested or outer.cancelled(), (
+        "the caller's cancellation was swallowed"
+    )
+
+
+@pytest.mark.parametrize("offset", OFFSETS)
+async def test_cancelling_run_forever_at_readiness_shuts_down(
+    docket: Docket, offset: int
+):
+    """run_forever cancelled the moment the worker becomes ready still exits"""
+    async with Worker(docket, name=f"cancel-me-{offset}") as worker:
+        run_task = asyncio.create_task(worker.run_forever())
+        await _wait_for_worker_readiness(worker)
+
+        for _ in range(offset):
+            await asyncio.sleep(0)
+        run_task.cancel()
+
+        _, pending = await asyncio.wait([run_task], timeout=5)
+
+        assert not pending, "the worker did not shut down after cancellation"