Fix Cancelled.__context__ leaking from other tasks

newsfragments/2649.bugfix.rst

@@ -0,0 +1,3 @@
+:exc:`Cancelled` exceptions no longer inherit ``__context__`` from
+unrelated exceptions being handled in the task that called
+:meth:`CancelScope.cancel`.

src/trio/_core/_run.py

@@ -1649,7 +1649,13 @@ def _attempt_abort(self, raise_cancel: _core.RaiseCancelT) -> None:
         # whether we succeeded or failed.
         self._abort_func = None
         if success is Abort.SUCCEEDED:
-            self._runner.reschedule(self, capture(raise_cancel))
+            # Clear __context__ to prevent exception state from the task
+            # that called cancel() from leaking into the cancelled task's
+            # Cancelled exception. See https://github.com/python-trio/trio/issues/2649
+            error = capture(raise_cancel)
+            error.error.__context__ = None
+            self._runner.reschedule(self, error)
+            del error
 
     def _attempt_delivery_of_any_pending_cancel(self) -> None:
         if self._abort_func is None:

src/trio/_core/_tests/test_run.py

@@ -603,6 +603,42 @@ async def sleeper() -> None:
                 assert record == ["sleeping", "cancelled"]
 
 
+async def test_cancelled_context_does_not_leak_from_other_task() -> None:
+    # https://github.com/python-trio/trio/issues/2649
+    # When task 1 calls cancel() while handling an exception,
+    # task 2's Cancelled exception should NOT have task 1's
+    # exception as __context__.
+    task2_cancelled: _core.Cancelled | None = None
+
+    async def task2(cancel_scope: _core.CancelScope) -> None:
+        nonlocal task2_cancelled
+        with cancel_scope:
+            try:
+                await sleep_forever()
+            except _core.Cancelled as exc:
+                task2_cancelled = exc
+                raise
+
+    async with _core.open_nursery() as nursery:
+        cancel_scope = _core.CancelScope()
+        nursery.start_soon(task2, cancel_scope)
+        await wait_all_tasks_blocked()
+
+        # Task 1 cancels the scope while handling an exception
+        try:
+            raise RuntimeError("task1 error")
+        except RuntimeError:
+            cancel_scope.cancel()
+
+        # Give task2 a chance to run and handle the cancellation
+        await wait_all_tasks_blocked()
+
+    assert task2_cancelled is not None
+    # The key assertion: task2's Cancelled should not have
+    # task1's RuntimeError as __context__
+    assert task2_cancelled.__context__ is None
+
+
 async def test_basic_timeout(mock_clock: _core.MockClock) -> None:
     start = _core.current_time()
     with _core.CancelScope() as scope: