gh-108668: Preserve exception state during delegated throw

Lib/test/test_yield_from.py

@@ -9,6 +9,8 @@
 
 import unittest
 import inspect
+import sys
+import types
 
 from test.support import captured_stderr, disable_gc, gc_collect
 from test import support
@@ -18,6 +20,83 @@ class TestPEP380Operation(unittest.TestCase):
     Test semantics.
     """
 
+    def test_delegated_throw_preserves_exception_state(self):
+        """Delegated throw inherits the delegating generator's exception."""
+        class Iterator:
+            def __init__(self, gen):
+                self.gen = gen
+
+            def __iter__(self):
+                return self
+
+            def __next__(self):
+                return next(self.gen)
+
+            def throw(self, *args):
+                return self.gen.throw(*args)
+
+        for wrap in (lambda gen: gen, Iterator):
+            with self.subTest(wrap=wrap):
+                original = RuntimeError("original")
+                seen = []
+
+                def inner():
+                    seen.append(sys.exception())
+                    try:
+                        yield
+                    except ValueError:
+                        pass
+                    seen.append(sys.exception())
+                    yield
+
+                def outer():
+                    try:
+                        raise original
+                    except RuntimeError:
+                        yield from wrap(inner())
+                        seen.append(sys.exception())
+
+                gen = outer()
+                next(gen)
+                try:
+                    raise LookupError("caller")
+                except LookupError as caller:
+                    gen.throw(ValueError())
+                    self.assertIs(sys.exception(), caller)
+                    with self.assertRaises(StopIteration):
+                        next(gen)
+                    self.assertIs(sys.exception(), caller)
+                self.assertEqual(seen, [original] * 3)
+
+    def test_await_throw_preserves_exception_state(self):
+        """Throwing through await preserves the surrounding handled exception."""
+        original = RuntimeError("original")
+        seen = []
+
+        @types.coroutine
+        def suspend():
+            yield
+
+        async def inner():
+            try:
+                await suspend()
+            except ValueError:
+                pass
+            seen.append(sys.exception())
+
+        async def outer():
+            try:
+                raise original
+            except RuntimeError:
+                await inner()
+                seen.append(sys.exception())
+
+        coro = outer()
+        coro.send(None)
+        with self.assertRaises(StopIteration):
+            coro.throw(ValueError())
+        self.assertEqual(seen, [original] * 2)
+
     def test_delegation_of_initial_next_to_subgenerator(self):
         """
         Test delegation of initial next() call to subgenerator

Misc/NEWS.d/next/Core_and_Builtins/2026-09-09-00-00-00.gh-issue-108668.1a2b3c.rst

@@ -0,0 +1,2 @@
+Preserve the handled exception state when delegating a generator's
+:meth:`~generator.throw` through ``yield from`` or ``await``.

Objects/genobject.c

@@ -682,8 +682,13 @@ _gen_throw(PyGenObject *gen, int close_on_genexit,
             tstate->current_frame = frame;
             /* Close the generator that we are currently iterating with
                'yield from' or awaiting on with 'await'. */
+            _PyErr_StackItem *prev_exc_info = tstate->exc_info;
+            gen->gi_exc_state.previous_item = prev_exc_info;
+            tstate->exc_info = &gen->gi_exc_state;
             ret = _gen_throw((PyGenObject *)yf, close_on_genexit,
                              typ, val, tb);
+            tstate->exc_info = prev_exc_info;
+            gen->gi_exc_state.previous_item = NULL;
             _PyThreadState_UpdateLastProfiledFrame(tstate, frame, prev);
             tstate->current_frame = prev;
             frame->previous = NULL;
@@ -704,7 +709,12 @@ _gen_throw(PyGenObject *gen, int close_on_genexit,
             _PyInterpreterFrame *prev = tstate->current_frame;
             frame->previous = prev;
             tstate->current_frame = frame;
+            _PyErr_StackItem *prev_exc_info = tstate->exc_info;
+            gen->gi_exc_state.previous_item = prev_exc_info;
+            tstate->exc_info = &gen->gi_exc_state;
             ret = PyObject_CallFunctionObjArgs(meth, typ, val, tb, NULL);
+            tstate->exc_info = prev_exc_info;
+            gen->gi_exc_state.previous_item = NULL;
             _PyThreadState_UpdateLastProfiledFrame(tstate, frame, prev);
             tstate->current_frame = prev;
             frame->previous = NULL;