diff --git a/Include/internal/pycore_crossinterp.h b/Include/internal/pycore_crossinterp.h
index ee9ff0090c2484..ac813cc499a008 100644
--- a/Include/internal/pycore_crossinterp.h
+++ b/Include/internal/pycore_crossinterp.h
@@ -164,17 +164,6 @@ extern void _PyXI_Fini(PyInterpreterState *interp);
 /* short-term data sharing */
 /***************************/
 
-// Ultimately we'd like to preserve enough information about the
-// exception and traceback that we could re-constitute (or at least
-// simulate, a la traceback.TracebackException), and even chain, a copy
-// of the exception in the calling interpreter.
-
-typedef struct _excinfo {
-    const char *type;
-    const char *msg;
-} _Py_excinfo;
-
-
 typedef enum error_code {
     _PyXI_ERR_NO_ERROR = 0,
     _PyXI_ERR_UNCAUGHT_EXCEPTION = -1,
@@ -197,9 +186,6 @@ typedef struct _sharedexception {
     _Py_excinfo uncaught;
 } _PyXI_exception_info;
 
-PyAPI_FUNC(void) _PyXI_ApplyExceptionInfo(
-    _PyXI_exception_info *info,
-    PyObject *exctype);
 
 typedef struct xi_session _PyXI_session;
 typedef struct _sharedns _PyXI_namespace;
@@ -269,6 +255,9 @@ PyAPI_FUNC(void) _PyXI_Exit(_PyXI_session *session);
 PyAPI_FUNC(void) _PyXI_ApplyCapturedException(
     _PyXI_session *session,
     PyObject *excwrapper);
+PyAPI_FUNC(PyObject *) _PyXI_ResolveCapturedException(
+    _PyXI_session *session,
+    PyObject *excwrapper);
 PyAPI_FUNC(int) _PyXI_HasCapturedException(_PyXI_session *session);
 
 
diff --git a/Include/internal/pycore_exceptions.h b/Include/internal/pycore_exceptions.h
index 4a9df709131998..4c9b3cbeb57169 100644
--- a/Include/internal/pycore_exceptions.h
+++ b/Include/internal/pycore_exceptions.h
@@ -8,16 +8,20 @@ extern "C" {
 #  error "this header requires Py_BUILD_CORE define"
 #endif
 
+#include "pycore_pyerrors.h"
+
 
 /* runtime lifecycle */
 
 extern PyStatus _PyExc_InitState(PyInterpreterState *);
 extern PyStatus _PyExc_InitGlobalObjects(PyInterpreterState *);
 extern int _PyExc_InitTypes(PyInterpreterState *);
+extern void _PyExc_FiniHeapObjects(PyInterpreterState *);
+extern void _PyExc_FiniTypes(PyInterpreterState *);
 extern void _PyExc_Fini(PyInterpreterState *);
 
 
-/* other API */
+/* runtime state */
 
 struct _Py_exc_state {
     // The dict mapping from errno codes to OSError subclasses
@@ -26,9 +30,17 @@ struct _Py_exc_state {
     int memerrors_numfree;
     // The ExceptionGroup type
     PyObject *PyExc_ExceptionGroup;
+
+    PyTypeObject *ExceptionSnapshotType;
 };
 
-extern void _PyExc_ClearExceptionGroupType(PyInterpreterState *);
+
+/* other API */
+
+PyAPI_FUNC(PyTypeObject *) _PyExc_GetExceptionSnapshotType(
+    PyInterpreterState *interp);
+
+extern PyObject * PyExceptionSnapshot_FromInfo(_Py_excinfo *info);
 
 
 #ifdef __cplusplus
diff --git a/Include/internal/pycore_pyerrors.h b/Include/internal/pycore_pyerrors.h
index 0f16fb894d17e1..a953d2bb18d4ad 100644
--- a/Include/internal/pycore_pyerrors.h
+++ b/Include/internal/pycore_pyerrors.h
@@ -68,6 +68,30 @@ extern PyStatus _PyErr_InitTypes(PyInterpreterState *);
 extern void _PyErr_FiniTypes(PyInterpreterState *);
 
 
+/* exception snapshots */
+
+// Ultimately we'd like to preserve enough information about the
+// exception and traceback that we could re-constitute (or at least
+// simulate, a la traceback.TracebackException), and even chain, a copy
+// of the exception in the calling interpreter.
+
+typedef struct _excinfo {
+    const char *type;
+    const char *msg;
+} _Py_excinfo;
+
+extern void _Py_excinfo_Clear(_Py_excinfo *info);
+extern int _Py_excinfo_Copy(_Py_excinfo *dest, _Py_excinfo *src);
+extern const char * _Py_excinfo_InitFromException(
+    _Py_excinfo *info,
+    PyObject *exc);
+extern void _Py_excinfo_Apply(_Py_excinfo *info, PyObject *exctype);
+extern const char * _Py_excinfo_AsUTF8(
+    _Py_excinfo *info,
+    char *buf,
+    size_t bufsize);
+
+
 /* other API */
 
 static inline PyObject* _PyErr_Occurred(PyThreadState *tstate)
diff --git a/Lib/test/support/interpreters.py b/Lib/test/support/interpreters.py
index ab9342b767dfae..b22c644068f039 100644
--- a/Lib/test/support/interpreters.py
+++ b/Lib/test/support/interpreters.py
@@ -14,6 +14,7 @@
 
 __all__ = [
     'Interpreter', 'get_current', 'get_main', 'create', 'list_all',
+    'RunFailedError',
     'SendChannel', 'RecvChannel',
     'create_channel', 'list_all_channels', 'is_shareable',
     'ChannelError', 'ChannelNotFoundError',
@@ -21,6 +22,17 @@
     ]
 
 
+class RunFailedError(RuntimeError):
+
+    def __init__(self, snapshot):
+        if snapshot.type and snapshot.msg:
+            msg = f'{snapshot.type}: {snapshot.msg}'
+        else:
+            msg = snapshot.type or snapshot.msg
+        super().__init__(msg)
+        self.snapshot = snapshot
+
+
 def create(*, isolated=True):
     """Return a new (idle) Python interpreter."""
     id = _interpreters.create(isolated=isolated)
@@ -91,14 +103,39 @@ def close(self):
         """
         return _interpreters.destroy(self._id)
 
+    # XXX setattr?
+    def bind(self, ns=None, /, **kwargs):
+        """Bind the given values into the interpreter's __main__.
+
+        The values must be shareable.
+        """
+        ns = dict(ns, **kwargs) if ns is not None else kwargs
+        _interpreters.set___main___attrs(self._id, ns)
+
+    # XXX getattr?
+    def get(self, name, default=None, /):
+        """Return the attr value from the interpreter's __main__.
+
+        The value must be shareable.
+        """
+        found = _interpreters.get___main___attrs(self._id, (name,), default)
+        assert len(found) == 1, found
+        return found[name]
+
     # XXX Rename "run" to "exec"?
-    def run(self, src_str, /, channels=None):
+    # XXX Do not allow init to overwrite (by default)?
+    def run(self, src_str, /, *, init=None):
         """Run the given source code in the interpreter.
 
         This is essentially the same as calling the builtin "exec"
         with this interpreter, using the __dict__ of its __main__
         module as both globals and locals.
 
+        If "init" is provided, it must be a dict mapping attribute names
+        to "shareable" objects, including channels.  These are set as
+        attributes on the __main__ module before the given code is
+        executed.  If a name is already bound then it is overwritten.
+
         There is no return value.
 
         If the code raises an unhandled exception then a RunFailedError
@@ -110,7 +147,9 @@ def run(self, src_str, /, channels=None):
         that time, the previous interpreter is allowed to run
         in other threads.
         """
-        _interpreters.exec(self._id, src_str, channels)
+        err = _interpreters.exec(self._id, src_str, init)
+        if err is not None:
+            raise RunFailedError(err)
 
 
 def create_channel():
diff --git a/Lib/test/test__xxinterpchannels.py b/Lib/test/test__xxinterpchannels.py
index 1c1ef3fac9d65f..03b0064f69eb7d 100644
--- a/Lib/test/test__xxinterpchannels.py
+++ b/Lib/test/test__xxinterpchannels.py
@@ -587,12 +587,12 @@ def test_run_string_arg_unresolved(self):
         cid = channels.create()
         interp = interpreters.create()
 
+        interpreters.set___main___attrs(interp, dict(cid=cid.send))
         out = _run_output(interp, dedent("""
             import _xxinterpchannels as _channels
             print(cid.end)
             _channels.send(cid, b'spam', blocking=False)
-            """),
-            dict(cid=cid.send))
+            """))
         obj = channels.recv(cid)
 
         self.assertEqual(obj, b'spam')
@@ -1017,16 +1017,16 @@ def test_close_multiple_users(self):
             _channels.recv({cid})
             """))
         channels.close(cid)
-        with self.assertRaises(interpreters.RunFailedError) as cm:
-            interpreters.run_string(id1, dedent(f"""
+
+        excsnap = interpreters.run_string(id1, dedent(f"""
                 _channels.send({cid}, b'spam')
                 """))
-        self.assertIn('ChannelClosedError', str(cm.exception))
-        with self.assertRaises(interpreters.RunFailedError) as cm:
-            interpreters.run_string(id2, dedent(f"""
+        self.assertIn('ChannelClosedError', excsnap.type)
+
+        excsnap = interpreters.run_string(id2, dedent(f"""
                 _channels.send({cid}, b'spam')
                 """))
-        self.assertIn('ChannelClosedError', str(cm.exception))
+        self.assertIn('ChannelClosedError', excsnap.type)
 
     def test_close_multiple_times(self):
         cid = channels.create()
diff --git a/Lib/test/test__xxsubinterpreters.py b/Lib/test/test__xxsubinterpreters.py
index ae7dfa19acc519..14df4d48c3f3ca 100644
--- a/Lib/test/test__xxsubinterpreters.py
+++ b/Lib/test/test__xxsubinterpreters.py
@@ -31,10 +31,10 @@ def _captured_script(script):
     return wrapped, open(r, encoding="utf-8")
 
 
-def _run_output(interp, request, shared=None):
+def _run_output(interp, request):
     script, rpipe = _captured_script(request)
     with rpipe:
-        interpreters.run_string(interp, script, shared)
+        interpreters.run_string(interp, script)
         return rpipe.read()
 
 
@@ -659,10 +659,10 @@ def test_shareable_types(self):
         ]
         for obj in objects:
             with self.subTest(obj):
+                interpreters.set___main___attrs(interp, dict(obj=obj))
                 interpreters.run_string(
                     interp,
                     f'assert(obj == {obj!r})',
-                    shared=dict(obj=obj),
                 )
 
     def test_os_exec(self):
@@ -743,30 +743,33 @@ def assert_run_failed(self, exctype, msg=None):
                              "{}: {}".format(exctype.__name__, msg))
 
     def test_invalid_syntax(self):
-        with self.assert_run_failed(SyntaxError):
-            # missing close paren
-            interpreters.run_string(self.id, 'print("spam"')
+        # missing close paren
+        exc = interpreters.run_string(self.id, 'print("spam"')
+        self.assertEqual(exc.type, 'SyntaxError')
 
     def test_failure(self):
-        with self.assert_run_failed(Exception, 'spam'):
-            interpreters.run_string(self.id, 'raise Exception("spam")')
+        exc = interpreters.run_string(self.id, 'raise Exception("spam")')
+        self.assertEqual(exc.type, 'Exception')
+        self.assertEqual(exc.msg, 'spam')
 
     def test_SystemExit(self):
-        with self.assert_run_failed(SystemExit, '42'):
-            interpreters.run_string(self.id, 'raise SystemExit(42)')
+        exc = interpreters.run_string(self.id, 'raise SystemExit(42)')
+        self.assertEqual(exc.type, 'SystemExit')
+        self.assertEqual(exc.msg, '42')
 
     def test_sys_exit(self):
-        with self.assert_run_failed(SystemExit):
-            interpreters.run_string(self.id, dedent("""
-                import sys
-                sys.exit()
-                """))
+        exc = interpreters.run_string(self.id, dedent("""
+            import sys
+            sys.exit()
+            """))
+        self.assertEqual(exc.type, 'SystemExit')
 
-        with self.assert_run_failed(SystemExit, '42'):
-            interpreters.run_string(self.id, dedent("""
-                import sys
-                sys.exit(42)
-                """))
+        exc = interpreters.run_string(self.id, dedent("""
+            import sys
+            sys.exit(42)
+            """))
+        self.assertEqual(exc.type, 'SystemExit')
+        self.assertEqual(exc.msg, '42')
 
     def test_with_shared(self):
         r, w = os.pipe()
@@ -787,7 +790,8 @@ def test_with_shared(self):
             with open({w}, 'wb') as chan:
                 pickle.dump(ns, chan)
             """)
-        interpreters.run_string(self.id, script, shared)
+        interpreters.set___main___attrs(self.id, shared)
+        interpreters.run_string(self.id, script)
         with open(r, 'rb') as chan:
             ns = pickle.load(chan)
 
@@ -808,7 +812,8 @@ def test_shared_overwrites(self):
             ns2 = dict(vars())
             del ns2['__builtins__']
         """)
-        interpreters.run_string(self.id, script, shared)
+        interpreters.set___main___attrs(self.id, shared)
+        interpreters.run_string(self.id, script)
 
         r, w = os.pipe()
         script = dedent(f"""
@@ -839,7 +844,8 @@ def test_shared_overwrites_default_vars(self):
             with open({w}, 'wb') as chan:
                 pickle.dump(ns, chan)
             """)
-        interpreters.run_string(self.id, script, shared)
+        interpreters.set___main___attrs(self.id, shared)
+        interpreters.run_string(self.id, script)
         with open(r, 'rb') as chan:
             ns = pickle.load(chan)
 
@@ -945,7 +951,8 @@ def script():
             with open(w, 'w', encoding="utf-8") as spipe:
                 with contextlib.redirect_stdout(spipe):
                     print('it worked!', end='')
-        interpreters.run_func(self.id, script, shared=dict(w=w))
+        interpreters.set___main___attrs(self.id, dict(w=w))
+        interpreters.run_func(self.id, script)
 
         with open(r, encoding="utf-8") as outfile:
             out = outfile.read()
@@ -961,7 +968,8 @@ def script():
                 with contextlib.redirect_stdout(spipe):
                     print('it worked!', end='')
         def f():
-            interpreters.run_func(self.id, script, shared=dict(w=w))
+            interpreters.set___main___attrs(self.id, dict(w=w))
+            interpreters.run_func(self.id, script)
         t = threading.Thread(target=f)
         t.start()
         t.join()
@@ -981,7 +989,8 @@ def script():
                 with contextlib.redirect_stdout(spipe):
                     print('it worked!', end='')
         code = script.__code__
-        interpreters.run_func(self.id, code, shared=dict(w=w))
+        interpreters.set___main___attrs(self.id, dict(w=w))
+        interpreters.run_func(self.id, code)
 
         with open(r, encoding="utf-8") as outfile:
             out = outfile.read()
diff --git a/Lib/test/test_import/__init__.py b/Lib/test/test_import/__init__.py
index aa465c70dfbcd0..1ecac4f37fe1c1 100644
--- a/Lib/test/test_import/__init__.py
+++ b/Lib/test/test_import/__init__.py
@@ -1968,10 +1968,12 @@ def test_disallowed_reimport(self):
             print(_testsinglephase)
             ''')
         interpid = _interpreters.create()
-        with self.assertRaises(_interpreters.RunFailedError):
-            _interpreters.run_string(interpid, script)
-        with self.assertRaises(_interpreters.RunFailedError):
-            _interpreters.run_string(interpid, script)
+
+        excsnap = _interpreters.run_string(interpid, script)
+        self.assertIsNot(excsnap, None)
+
+        excsnap = _interpreters.run_string(interpid, script)
+        self.assertIsNot(excsnap, None)
 
 
 class TestSinglePhaseSnapshot(ModuleSnapshot):
diff --git a/Lib/test/test_importlib/test_util.py b/Lib/test/test_importlib/test_util.py
index 5da72a21f586ee..f03f6677bbd913 100644
--- a/Lib/test/test_importlib/test_util.py
+++ b/Lib/test/test_importlib/test_util.py
@@ -655,25 +655,19 @@ def test_magic_number(self):
 @unittest.skipIf(_interpreters is None, 'subinterpreters required')
 class IncompatibleExtensionModuleRestrictionsTests(unittest.TestCase):
 
-    ERROR = re.compile("^ImportError: module (.*) does not support loading in subinterpreters")
-
     def run_with_own_gil(self, script):
         interpid = _interpreters.create(isolated=True)
-        try:
-            _interpreters.run_string(interpid, script)
-        except _interpreters.RunFailedError as exc:
-            if m := self.ERROR.match(str(exc)):
-                modname, = m.groups()
-                raise ImportError(modname)
+        excsnap = _interpreters.run_string(interpid, script)
+        if excsnap is not None:
+            if excsnap.type == 'ImportError':
+                raise ImportError(excsnap.msg)
 
     def run_with_shared_gil(self, script):
         interpid = _interpreters.create(isolated=False)
-        try:
-            _interpreters.run_string(interpid, script)
-        except _interpreters.RunFailedError as exc:
-            if m := self.ERROR.match(str(exc)):
-                modname, = m.groups()
-                raise ImportError(modname)
+        excsnap = _interpreters.run_string(interpid, script)
+        if excsnap is not None:
+            if excsnap.type == 'ImportError':
+                raise ImportError(excsnap.msg)
 
     @unittest.skipIf(_testsinglephase is None, "test requires _testsinglephase module")
     def test_single_phase_init_module(self):
diff --git a/Lib/test/test_interpreters.py b/Lib/test/test_interpreters.py
index 74f86088b45590..80cb0783f76243 100644
--- a/Lib/test/test_interpreters.py
+++ b/Lib/test/test_interpreters.py
@@ -39,10 +39,12 @@ def clean_up_interpreters():
             pass  # already destroyed
 
 
-def _run_output(interp, request, channels=None):
+def _run_output(interp, request, init=None):
     script, rpipe = _captured_script(request)
     with rpipe:
-        interp.run(script, channels=channels)
+        if init:
+            interp.bind(init)
+        interp.run(script)
         return rpipe.read()
 
 
@@ -467,6 +469,93 @@ def task():
         self.assertEqual(os.read(r_interp, 1), FINISHED)
 
 
+class TestInterpreterBind(TestBase):
+
+    def test_empty(self):
+        interp = interpreters.create()
+        with self.assertRaises(ValueError):
+            interp.bind()
+
+    def test_dict(self):
+        values = {'spam': 42, 'eggs': 'ham'}
+        interp = interpreters.create()
+        interp.bind(values)
+        out = _run_output(interp, dedent("""
+            print(spam, eggs)
+            """))
+        self.assertEqual(out.strip(), '42 ham')
+
+    def test_tuple(self):
+        values = {'spam': 42, 'eggs': 'ham'}
+        values = tuple(values.items())
+        interp = interpreters.create()
+        interp.bind(values)
+        out = _run_output(interp, dedent("""
+            print(spam, eggs)
+            """))
+        self.assertEqual(out.strip(), '42 ham')
+
+    def test_kwargs(self):
+        values = {'spam': 42, 'eggs': 'ham'}
+        interp = interpreters.create()
+        interp.bind(**values)
+        out = _run_output(interp, dedent("""
+            print(spam, eggs)
+            """))
+        self.assertEqual(out.strip(), '42 ham')
+
+    def test_dict_and_kwargs(self):
+        values = {'spam': 42, 'eggs': 'ham'}
+        interp = interpreters.create()
+        interp.bind(values, foo='bar')
+        out = _run_output(interp, dedent("""
+            print(spam, eggs, foo)
+            """))
+        self.assertEqual(out.strip(), '42 ham bar')
+
+    def test_not_shareable(self):
+        interp = interpreters.create()
+        # XXX TypeError?
+        with self.assertRaises(ValueError):
+            interp.bind(spam={'spam': 'eggs', 'foo': 'bar'})
+
+        # Make sure neither was actually bound.
+        with self.assertRaises(RuntimeError):
+            interp.run('print(foo)')
+        with self.assertRaises(RuntimeError):
+            interp.run('print(spam)')
+
+
+class TestInterpreterGet(TestBase):
+
+    def test_empty(self):
+        interp = interpreters.create()
+        with self.assertRaises(TypeError):
+            interp.get()
+
+    def test_found(self):
+        interp = interpreters.create()
+        obj1 = interp.get('__name__')
+        interp.bind(spam=42)
+        obj2 = interp.get('spam')
+
+        self.assertEqual(obj1, '__main__')
+        self.assertEqual(obj2, 42)
+
+    def test_not_found(self):
+        interp = interpreters.create()
+        obj1 = interp.get('spam')
+        obj2 = interp.get('spam', 'eggs')
+
+        self.assertIs(obj1, None)
+        self.assertEqual(obj2, 'eggs')
+
+    def test_not_shareable(self):
+        interp = interpreters.create()
+        with self.assertRaises(ValueError):
+            interp.get('__builtins__')
+
+
 class TestInterpreterRun(TestBase):
 
     def test_success(self):
@@ -478,6 +567,11 @@ def test_success(self):
 
         self.assertEqual(out, 'it worked!')
 
+    def test_failure(self):
+        interp = interpreters.create()
+        with self.assertRaises(interpreters.RunFailedError):
+            interp.run('raise Exception')
+
     def test_in_thread(self):
         interp = interpreters.create()
         script, file = _captured_script('print("it worked!", end="")')
@@ -953,7 +1047,7 @@ def test_send_recv_different_interpreters(self):
                 print(id(orig2))
                 s.send_nowait(orig2)
                 """),
-            channels=dict(r=r1, s=s2),
+            init=dict(r=r1, s=s2),
             )
         obj2 = r2.recv()
 
@@ -1027,7 +1121,7 @@ def test_send_recv_nowait_different_interpreters(self):
                 print(id(orig2))
                 s.send_nowait(orig2)
                 """),
-            channels=dict(r=r1, s=s2),
+            init=dict(r=r1, s=s2),
             )
         obj2 = r2.recv_nowait()
 
diff --git a/Modules/_xxsubinterpretersmodule.c b/Modules/_xxsubinterpretersmodule.c
index 001fa887847cbd..f30c1d351eba02 100644
--- a/Modules/_xxsubinterpretersmodule.c
+++ b/Modules/_xxsubinterpretersmodule.c
@@ -7,7 +7,7 @@
 
 #include "Python.h"
 #include "pycore_crossinterp.h"   // struct _xid
-#include "pycore_pyerrors.h"      // _Py_excinfo
+#include "pycore_exceptions.h"    // PyExceptionSnapshot_FromInfo()
 #include "pycore_initconfig.h"    // _PyErr_SetFromPyStatus()
 #include "pycore_modsupport.h"    // _PyArg_BadArgument()
 #include "pycore_pyerrors.h"      // _PyErr_ChainExceptions1()
@@ -28,31 +28,12 @@ _get_current_interp(void)
     return PyInterpreterState_Get();
 }
 
-static PyObject *
-add_new_exception(PyObject *mod, const char *name, PyObject *base)
-{
-    assert(!PyObject_HasAttrStringWithError(mod, name));
-    PyObject *exctype = PyErr_NewException(name, base, NULL);
-    if (exctype == NULL) {
-        return NULL;
-    }
-    int res = PyModule_AddType(mod, (PyTypeObject *)exctype);
-    if (res < 0) {
-        Py_DECREF(exctype);
-        return NULL;
-    }
-    return exctype;
-}
-
-#define ADD_NEW_EXCEPTION(MOD, NAME, BASE) \
-    add_new_exception(MOD, MODULE_NAME "." Py_STRINGIFY(NAME), BASE)
-
 
 /* module state *************************************************************/
 
 typedef struct {
-    /* exceptions */
-    PyObject *RunFailedError;
+    /* heap types */
+    PyTypeObject *ExceptionSnapshotType;
 } module_state;
 
 static inline module_state *
@@ -67,8 +48,8 @@ get_module_state(PyObject *mod)
 static int
 traverse_module_state(module_state *state, visitproc visit, void *arg)
 {
-    /* exceptions */
-    Py_VISIT(state->RunFailedError);
+    /* heap types */
+    Py_VISIT(state->ExceptionSnapshotType);
 
     return 0;
 }
@@ -76,8 +57,8 @@ traverse_module_state(module_state *state, visitproc visit, void *arg)
 static int
 clear_module_state(module_state *state)
 {
-    /* exceptions */
-    Py_CLEAR(state->RunFailedError);
+    /* heap types */
+    Py_CLEAR(state->ExceptionSnapshotType);
 
     return 0;
 }
@@ -177,30 +158,6 @@ get_code_str(PyObject *arg, Py_ssize_t *len_p, PyObject **bytes_p, int *flags_p)
 
 /* interpreter-specific code ************************************************/
 
-static int
-exceptions_init(PyObject *mod)
-{
-    module_state *state = get_module_state(mod);
-    if (state == NULL) {
-        return -1;
-    }
-
-#define ADD(NAME, BASE) \
-    do { \
-        assert(state->NAME == NULL); \
-        state->NAME = ADD_NEW_EXCEPTION(mod, NAME, BASE); \
-        if (state->NAME == NULL) { \
-            return -1; \
-        } \
-    } while (0)
-
-    // An uncaught exception came out of interp_run_string().
-    ADD(RunFailedError, PyExc_RuntimeError);
-#undef ADD
-
-    return 0;
-}
-
 static int
 _run_script(PyObject *ns, const char *codestr, Py_ssize_t codestrlen, int flags)
 {
@@ -228,8 +185,7 @@ _run_script(PyObject *ns, const char *codestr, Py_ssize_t codestrlen, int flags)
 static int
 _run_in_interpreter(PyInterpreterState *interp,
                     const char *codestr, Py_ssize_t codestrlen,
-                    PyObject *shareables, int flags,
-                    PyObject *excwrapper)
+                    PyObject *shareables, int flags, PyObject **excsnap)
 {
     assert(!PyErr_Occurred());
     _PyXI_session session = {0};
@@ -237,9 +193,9 @@ _run_in_interpreter(PyInterpreterState *interp,
     // Prep and switch interpreters.
     if (_PyXI_Enter(&session, interp, shareables) < 0) {
         assert(!PyErr_Occurred());
-        _PyXI_ApplyExceptionInfo(session.exc, excwrapper);
-        assert(PyErr_Occurred());
-        return -1;
+        *excsnap = _PyXI_ResolveCapturedException(&session, NULL);
+        assert((PyErr_Occurred() == NULL) != (*excsnap == NULL));
+        return PyErr_Occurred() ? -1 : 0;
     }
 
     // Run the script.
@@ -249,9 +205,12 @@ _run_in_interpreter(PyInterpreterState *interp,
     _PyXI_Exit(&session);
 
     // Propagate any exception out to the caller.
-    assert(!PyErr_Occurred());
     if (res < 0) {
-        _PyXI_ApplyCapturedException(&session, excwrapper);
+        *excsnap = _PyXI_ResolveCapturedException(&session, NULL);
+        assert((PyErr_Occurred() == NULL) != (*excsnap == NULL));
+        if (!PyErr_Occurred()) {
+            res = 0;
+        }
     }
     else {
         assert(!_PyXI_HasCapturedException(&session));
@@ -443,6 +402,138 @@ PyDoc_STRVAR(get_main_doc,
 \n\
 Return the ID of main interpreter.");
 
+static PyObject *
+interp_set___main___attrs(PyObject *self, PyObject *args)
+{
+    PyObject *id, *updates;
+    if (!PyArg_ParseTuple(args, "OO:" MODULE_NAME ".set___main___attrs",
+                          &id, &updates))
+    {
+        return NULL;
+    }
+
+    // Look up the interpreter.
+    PyInterpreterState *interp = PyInterpreterID_LookUp(id);
+    if (interp == NULL) {
+        return NULL;
+    }
+
+    // Check the updates.
+    if (updates != Py_None) {
+        Py_ssize_t size = PyObject_Size(updates);
+        if (size < 0) {
+            return NULL;
+        }
+        if (size == 0) {
+            PyErr_SetString(PyExc_ValueError,
+                            "arg 2 must be a non-empty mapping");
+            return NULL;
+        }
+    }
+
+    _PyXI_session session = {0};
+
+    // Prep and switch interpreters, including apply the updates.
+    if (_PyXI_Enter(&session, interp, updates) < 0) {
+        if (!PyErr_Occurred()) {
+            _PyXI_ApplyCapturedException(&session, NULL);
+            assert(PyErr_Occurred());
+        }
+        else {
+            assert(!_PyXI_HasCapturedException(&session));
+        }
+        return NULL;
+    }
+
+    // Clean up and switch back.
+    _PyXI_Exit(&session);
+
+    Py_RETURN_NONE;
+}
+
+PyDoc_STRVAR(set___main___attrs_doc,
+"set___main___attrs(id, ns)\n\
+\n\
+Bind the given attributes in the interpreter's __main__ module.");
+
+static PyObject *
+interp_get___main___attrs(PyObject *self, PyObject *args)
+{
+    PyObject *id, *names;
+    PyObject *dflt = Py_None;
+    if (!PyArg_ParseTuple(args, "OO|O:" MODULE_NAME ".get___main___attrs",
+                          &id, &names, &dflt))
+    {
+        return NULL;
+    }
+
+    // Look up the interpreter.
+    PyInterpreterState *interp = PyInterpreterID_LookUp(id);
+    if (interp == NULL) {
+        return NULL;
+    }
+
+    // Prep the result.
+    PyObject *found = PyDict_New();
+    if (found == NULL) {
+        return NULL;
+    }
+
+    // Set up the shared ns.
+    _PyXI_namespace *shared = _PyXI_NamespaceFromNames(names);
+    if (shared == NULL) {
+        if (!PyErr_Occurred()) {
+            PyErr_SetString(PyExc_ValueError, "expected non-empty list of names");
+        }
+        Py_DECREF(found);
+        return NULL;
+    }
+
+    _PyXI_session session = {0};
+
+    // Prep and switch interpreters, including apply the updates.
+    if (_PyXI_Enter(&session, interp, NULL) < 0) {
+        Py_DECREF(found);
+        assert(!PyErr_Occurred());
+        _PyXI_ApplyCapturedException(&session, NULL);
+        assert(PyErr_Occurred());
+        return NULL;
+    }
+
+    // Extract the requested attrs from __main__.
+    int res = _PyXI_FillNamespaceFromDict(shared, session.main_ns, &session);
+
+    // Clean up and switch back.
+    _PyXI_Exit(&session);
+
+    if (res == 0) {
+        assert(!PyErr_Occurred());
+        // Copy the objects into the result dict.
+        if (_PyXI_ApplyNamespace(shared, found, dflt) < 0) {
+            Py_CLEAR(found);
+        }
+    }
+    else {
+        if (!PyErr_Occurred()) {
+            _PyXI_ApplyCapturedException(&session, NULL);
+            assert(PyErr_Occurred());
+        }
+        else {
+            assert(!_PyXI_HasCapturedException(&session));
+        }
+        Py_CLEAR(found);
+    }
+
+    _PyXI_FreeNamespace(shared);
+    return found;
+}
+
+PyDoc_STRVAR(get___main___attrs_doc,
+"get___main___attrs(id, names, default=None, /)\n\
+\n\
+Look up the given attributes in the interpreter's __main__ module.\n\
+Return the default if not found.");
+
 static PyUnicodeObject *
 convert_script_arg(PyObject *arg, const char *fname, const char *displayname,
                    const char *expected)
@@ -519,14 +610,15 @@ convert_code_arg(PyObject *arg, const char *fname, const char *displayname,
     return code;
 }
 
-static int
+static PyObject *
 _interp_exec(PyObject *self,
              PyObject *id_arg, PyObject *code_arg, PyObject *shared_arg)
 {
     // Look up the interpreter.
     PyInterpreterState *interp = PyInterpreterID_LookUp(id_arg);
     if (interp == NULL) {
-        return -1;
+        assert(PyErr_Occurred());
+        return NULL;
     }
 
     // Extract code.
@@ -536,20 +628,24 @@ _interp_exec(PyObject *self,
     const char *codestr = get_code_str(code_arg,
                                        &codestrlen, &bytes_obj, &flags);
     if (codestr == NULL) {
-        return -1;
+        assert(PyErr_Occurred());
+        return NULL;
     }
 
     // Run the code in the interpreter.
-    module_state *state = get_module_state(self);
-    assert(state != NULL);
+    PyObject *excsnap = NULL;
     int res = _run_in_interpreter(interp, codestr, codestrlen,
-                                  shared_arg, flags, state->RunFailedError);
+                                  shared_arg, flags, &excsnap);
     Py_XDECREF(bytes_obj);
     if (res < 0) {
-        return -1;
+        assert(PyErr_Occurred());
+        assert(excsnap == NULL);
+        return NULL;
     }
-
-    return 0;
+    else if (excsnap != NULL) {
+        return excsnap;
+    }
+    Py_RETURN_NONE;
 }
 
 static PyObject *
@@ -577,12 +673,9 @@ interp_exec(PyObject *self, PyObject *args, PyObject *kwds)
         return NULL;
     }
 
-    int res = _interp_exec(self, id, code, shared);
+    PyObject *res = _interp_exec(self, id, code, shared);
     Py_DECREF(code);
-    if (res < 0) {
-        return NULL;
-    }
-    Py_RETURN_NONE;
+    return res;
 }
 
 PyDoc_STRVAR(exec_doc,
@@ -620,12 +713,9 @@ interp_run_string(PyObject *self, PyObject *args, PyObject *kwds)
         return NULL;
     }
 
-    int res = _interp_exec(self, id, script, shared);
+    PyObject *res = _interp_exec(self, id, script, shared);
     Py_DECREF(script);
-    if (res < 0) {
-        return NULL;
-    }
-    Py_RETURN_NONE;
+    return res;
 }
 
 PyDoc_STRVAR(run_string_doc,
@@ -654,12 +744,9 @@ interp_run_func(PyObject *self, PyObject *args, PyObject *kwds)
         return NULL;
     }
 
-    int res = _interp_exec(self, id, (PyObject *)code, shared);
+    PyObject *res = _interp_exec(self, id, (PyObject *)code, shared);
     Py_DECREF(code);
-    if (res < 0) {
-        return NULL;
-    }
-    Py_RETURN_NONE;
+    return res;
 }
 
 PyDoc_STRVAR(run_func_doc,
@@ -743,6 +830,10 @@ static PyMethodDef module_functions[] = {
     {"run_func",                  _PyCFunction_CAST(interp_run_func),
      METH_VARARGS | METH_KEYWORDS, run_func_doc},
 
+    {"set___main___attrs",        _PyCFunction_CAST(interp_set___main___attrs),
+     METH_VARARGS, set___main___attrs_doc},
+    {"get___main___attrs",        _PyCFunction_CAST(interp_get___main___attrs),
+     METH_VARARGS, get___main___attrs_doc},
     {"is_shareable",              _PyCFunction_CAST(object_is_shareable),
      METH_VARARGS | METH_KEYWORDS, is_shareable_doc},
 
@@ -759,8 +850,9 @@ The 'interpreters' module provides a more convenient interface.");
 static int
 module_exec(PyObject *mod)
 {
-    /* Add exception types */
-    if (exceptions_init(mod) != 0) {
+    PyInterpreterState *interp = PyInterpreterState_Get();
+    module_state *state = get_module_state(mod);
+    if (state == NULL) {
         goto error;
     }
 
@@ -769,6 +861,12 @@ module_exec(PyObject *mod)
         goto error;
     }
 
+    // ExceptionSnapshot
+    state->ExceptionSnapshotType = _PyExc_GetExceptionSnapshotType(interp);
+    if (PyModule_AddType(mod, state->ExceptionSnapshotType) < 0) {
+        goto error;
+    }
+
     return 0;
 
 error:
diff --git a/Objects/exceptions.c b/Objects/exceptions.c
index a685ed803cd02d..13867bbb8b944e 100644
--- a/Objects/exceptions.c
+++ b/Objects/exceptions.c
@@ -26,9 +26,8 @@ PyObject *PyExc_WindowsError = NULL;  // borrowed ref
 
 
 static struct _Py_exc_state*
-get_exc_state(void)
+get_exc_state(PyInterpreterState *interp)
 {
-    PyInterpreterState *interp = _PyInterpreterState_GET();
     return &interp->exc_state;
 }
 
@@ -697,7 +696,8 @@ _PyBaseExceptionGroupObject_cast(PyObject *exc)
 static PyObject *
 BaseExceptionGroup_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
 {
-    struct _Py_exc_state *state = get_exc_state();
+    PyInterpreterState *interp = _PyInterpreterState_GET();
+    struct _Py_exc_state *state = get_exc_state(interp);
     PyTypeObject *PyExc_ExceptionGroup =
         (PyTypeObject*)state->PyExc_ExceptionGroup;
 
@@ -1491,7 +1491,8 @@ ComplexExtendsException(PyExc_BaseException, BaseExceptionGroup,
  */
 static PyObject*
 create_exception_group_class(void) {
-    struct _Py_exc_state *state = get_exc_state();
+    PyInterpreterState *interp = _PyInterpreterState_GET();
+    struct _Py_exc_state *state = get_exc_state(interp);
 
     PyObject *bases = PyTuple_Pack(
         2, PyExc_BaseExceptionGroup, PyExc_Exception);
@@ -1858,7 +1859,8 @@ OSError_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
             ))
             goto error;
 
-        struct _Py_exc_state *state = get_exc_state();
+        PyInterpreterState *interp = _PyInterpreterState_GET();
+        struct _Py_exc_state *state = get_exc_state(interp);
         if (myerrno && PyLong_Check(myerrno) &&
             state->errnomap && (PyObject *) type == PyExc_OSError) {
             PyObject *newtype;
@@ -3283,7 +3285,8 @@ static PyObject *
 get_memory_error(int allow_allocation, PyObject *args, PyObject *kwds)
 {
     PyBaseExceptionObject *self;
-    struct _Py_exc_state *state = get_exc_state();
+    PyInterpreterState *interp = _PyInterpreterState_GET();
+    struct _Py_exc_state *state = get_exc_state(interp);
     if (state->memerrors_freelist == NULL) {
         if (!allow_allocation) {
             PyInterpreterState *interp = _PyInterpreterState_GET();
@@ -3352,7 +3355,8 @@ MemoryError_dealloc(PyBaseExceptionObject *self)
         return;
     }
 
-    struct _Py_exc_state *state = get_exc_state();
+    PyInterpreterState *interp = _PyInterpreterState_GET();
+    struct _Py_exc_state *state = get_exc_state(interp);
     if (state->memerrors_numfree >= MEMERRORS_SAVE) {
         Py_TYPE(self)->tp_free((PyObject *)self);
     }
@@ -3660,6 +3664,9 @@ static struct static_exception static_exceptions[] = {
 };
 
 
+static int
+_exc_snapshot_init_type(PyInterpreterState *interp);
+
 int
 _PyExc_InitTypes(PyInterpreterState *interp)
 {
@@ -3669,11 +3676,13 @@ _PyExc_InitTypes(PyInterpreterState *interp)
             return -1;
         }
     }
+    if (_exc_snapshot_init_type(interp) < 0) {
+        return -1;
+    }
     return 0;
 }
 
-
-static void
+void
 _PyExc_FiniTypes(PyInterpreterState *interp)
 {
     for (Py_ssize_t i=Py_ARRAY_LENGTH(static_exceptions) - 1; i >= 0; i--) {
@@ -3792,11 +3801,16 @@ _PyBuiltins_AddExceptions(PyObject *bltinmod)
     return 0;
 }
 
+
+// _PyExc_FiniHeapObjects() must be called before the interpreter
+// state is cleared, since there are heap types to clean up.
+
 void
-_PyExc_ClearExceptionGroupType(PyInterpreterState *interp)
+_PyExc_FiniHeapObjects(PyInterpreterState *interp)
 {
-    struct _Py_exc_state *state = &interp->exc_state;
+    struct _Py_exc_state *state = get_exc_state(interp);
     Py_CLEAR(state->PyExc_ExceptionGroup);
+    Py_CLEAR(state->ExceptionSnapshotType);
 }
 
 void
@@ -3805,8 +3819,6 @@ _PyExc_Fini(PyInterpreterState *interp)
     struct _Py_exc_state *state = &interp->exc_state;
     free_preallocated_memerrors(state);
     Py_CLEAR(state->errnomap);
-
-    _PyExc_FiniTypes(interp);
 }
 
 int
@@ -3824,3 +3836,175 @@ _PyException_AddNote(PyObject *exc, PyObject *note)
     return res;
 }
 
+
+/* exception snapshots */
+
+typedef struct exc_snapshot {
+    PyObject_HEAD
+    _Py_excinfo info;
+} PyExceptionSnapshotObject;
+
+static void
+exc_snapshot_dealloc(PyExceptionSnapshotObject *self)
+{
+    PyTypeObject *tp = Py_TYPE(self);
+    _Py_excinfo_Clear(&self->info);
+    tp->tp_free(self);
+    /* "Instances of heap-allocated types hold a reference to their type."
+     * See: https://docs.python.org/3.11/howto/isolating-extensions.html#garbage-collection-protocol
+     * See: https://docs.python.org/3.11/c-api/typeobj.html#c.PyTypeObject.tp_traverse
+    */
+    // XXX Why don't we implement Py_TPFLAGS_HAVE_GC, e.g. Py_tp_traverse,
+    // like we do for _abc._abc_data?
+    Py_DECREF(tp);
+}
+
+static PyObject *
+exc_snapshot_repr(PyExceptionSnapshotObject *self)
+{
+    PyTypeObject *type = Py_TYPE(self);
+    const char *clsname = _PyType_Name(type);
+    return PyUnicode_FromFormat("%s(name='%s', msg='%s')",
+                                clsname, self->info.type, self->info.msg);
+}
+
+static PyObject *
+exc_snapshot_str(PyExceptionSnapshotObject *self)
+{
+    char buf[256];
+    const char *msg = _Py_excinfo_AsUTF8(&self->info, buf, 256);
+    if (msg == NULL) {
+        msg = "";
+    }
+    return PyUnicode_FromString(msg);
+}
+
+static Py_hash_t
+exc_snapshot_hash(PyExceptionSnapshotObject *self)
+{
+    PyObject *str = exc_snapshot_str(self);
+    if (str == NULL) {
+        return -1;
+    }
+    Py_hash_t hash = PyObject_Hash(str);
+    Py_DECREF(str);
+    return hash;
+}
+
+PyDoc_STRVAR(exc_snapshot_doc,
+"ExceptionSnapshot\n\
+\n\
+A minimal summary of a raised exception.");
+
+static PyMemberDef exc_snapshot_members[] = {
+#define OFFSET(field) \
+        (offsetof(PyExceptionSnapshotObject, info) + offsetof(_Py_excinfo, field))
+    {"type", Py_T_STRING, OFFSET(type), Py_READONLY,
+     PyDoc_STR("the name of the original exception type")},
+    {"msg", Py_T_STRING, OFFSET(msg), Py_READONLY,
+     PyDoc_STR("the message string of the original exception")},
+#undef OFFSET
+    {NULL}
+};
+
+static PyObject *
+exc_snapshot_apply(PyExceptionSnapshotObject *self,
+                   PyObject *args, PyObject *kwargs)
+{
+    static char *kwlist[] = {"exctype", NULL};
+    PyObject *exctype = NULL;
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs,
+                                     "|O:ExceptionSnapshot.apply" , kwlist,
+                                     &exctype)) {
+        return NULL;
+    }
+
+    if (exctype == NULL) {
+        exctype = PyExc_RuntimeError;
+    }
+
+    _Py_excinfo_Apply(&self->info, exctype);
+    return NULL;
+}
+
+PyDoc_STRVAR(exc_snapshot_apply_doc,
+"Raise an exception based on the snapshot.");
+
+static PyMethodDef exc_snapshot_methods[] = {
+    {"apply",                    _PyCFunction_CAST(exc_snapshot_apply),
+     METH_VARARGS | METH_KEYWORDS, exc_snapshot_apply_doc},
+    {NULL}
+};
+
+static PyType_Slot ExcSnapshotType_slots[] = {
+    {Py_tp_dealloc, (destructor)exc_snapshot_dealloc},
+    {Py_tp_doc, (void *)exc_snapshot_doc},
+    {Py_tp_repr, (reprfunc)exc_snapshot_repr},
+    {Py_tp_str, (reprfunc)exc_snapshot_str},
+    {Py_tp_hash, exc_snapshot_hash},
+    {Py_tp_members, exc_snapshot_members},
+    {Py_tp_methods, exc_snapshot_methods},
+    {0, NULL},
+};
+
+static PyType_Spec ExcSnapshotType_spec = {
+    // XXX Move it to builtins?
+    .name = "_interpreters.ExceptionSnapshot",
+    .basicsize = sizeof(PyExceptionSnapshotObject),
+    .flags = (Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE |
+              Py_TPFLAGS_DISALLOW_INSTANTIATION | Py_TPFLAGS_IMMUTABLETYPE),
+    .slots = ExcSnapshotType_slots,
+};
+
+static int
+_exc_snapshot_init_type(PyInterpreterState *interp)
+{
+    struct _Py_exc_state *state = get_exc_state(interp);
+    assert(state->ExceptionSnapshotType == NULL);
+    PyTypeObject *cls = (PyTypeObject *)PyType_FromMetaclass(
+                NULL, NULL, &ExcSnapshotType_spec, NULL);
+    if (cls == NULL) {
+        return -1;
+    }
+    state->ExceptionSnapshotType = cls;
+    return 0;
+}
+
+PyTypeObject *
+_PyExc_GetExceptionSnapshotType(PyInterpreterState *interp)
+{
+    struct _Py_exc_state *state = get_exc_state(interp);
+    assert(state->ExceptionSnapshotType != NULL);
+    return (PyTypeObject *)Py_NewRef(state->ExceptionSnapshotType);
+}
+
+static PyExceptionSnapshotObject *
+new_exc_snapshot(PyInterpreterState *interp)
+{
+    struct _Py_exc_state *state = get_exc_state(interp);
+    assert(state->ExceptionSnapshotType != NULL);
+    PyTypeObject *cls = state->ExceptionSnapshotType;
+
+    PyExceptionSnapshotObject *self = \
+        (PyExceptionSnapshotObject *)PyObject_New(PyExceptionSnapshotObject, cls);
+    if (self == NULL) {
+        PyErr_NoMemory();
+        return NULL;
+    }
+    self->info = (_Py_excinfo){0};
+    return self;
+}
+
+PyObject *
+PyExceptionSnapshot_FromInfo(_Py_excinfo *info)
+{
+    PyInterpreterState *interp = _PyInterpreterState_GET();
+    PyExceptionSnapshotObject *self = new_exc_snapshot(interp);
+    if (self == NULL) {
+        return NULL;
+    }
+    if (_Py_excinfo_Copy(&self->info, info) < 0) {
+        Py_DECREF(self);
+    }
+    return (PyObject *)self;
+}
diff --git a/Python/crossinterp.c b/Python/crossinterp.c
index de28cb7071740a..e4caa6c12cc3a0 100644
--- a/Python/crossinterp.c
+++ b/Python/crossinterp.c
@@ -800,17 +800,6 @@ _xidregistry_fini(struct _xidregistry *registry)
 /* convenience utilities */
 /*************************/
 
-static const char *
-_copy_raw_string(const char *str)
-{
-    char *copied = PyMem_RawMalloc(strlen(str)+1);
-    if (copied == NULL) {
-        return NULL;
-    }
-    strcpy(copied, str);
-    return copied;
-}
-
 static const char *
 _copy_string_obj_raw(PyObject *strobj)
 {
@@ -846,118 +835,6 @@ _release_xid_data(_PyCrossInterpreterData *data, int rawfree)
 }
 
 
-/* exception snapshots */
-
-static int
-_exc_type_name_as_utf8(PyObject *exc, const char **p_typename)
-{
-    // XXX Use PyObject_GetAttrString(Py_TYPE(exc), '__name__')?
-    PyObject *nameobj = PyUnicode_FromString(Py_TYPE(exc)->tp_name);
-    if (nameobj == NULL) {
-        assert(PyErr_Occurred());
-        *p_typename = "unable to format exception type name";
-        return -1;
-    }
-    const char *name = PyUnicode_AsUTF8(nameobj);
-    if (name == NULL) {
-        assert(PyErr_Occurred());
-        Py_DECREF(nameobj);
-        *p_typename = "unable to encode exception type name";
-        return -1;
-    }
-    name = _copy_raw_string(name);
-    Py_DECREF(nameobj);
-    if (name == NULL) {
-        *p_typename = "out of memory copying exception type name";
-        return -1;
-    }
-    *p_typename = name;
-    return 0;
-}
-
-static int
-_exc_msg_as_utf8(PyObject *exc, const char **p_msg)
-{
-    PyObject *msgobj = PyObject_Str(exc);
-    if (msgobj == NULL) {
-        assert(PyErr_Occurred());
-        *p_msg = "unable to format exception message";
-        return -1;
-    }
-    const char *msg = PyUnicode_AsUTF8(msgobj);
-    if (msg == NULL) {
-        assert(PyErr_Occurred());
-        Py_DECREF(msgobj);
-        *p_msg = "unable to encode exception message";
-        return -1;
-    }
-    msg = _copy_raw_string(msg);
-    Py_DECREF(msgobj);
-    if (msg == NULL) {
-        assert(PyErr_ExceptionMatches(PyExc_MemoryError));
-        *p_msg = "out of memory copying exception message";
-        return -1;
-    }
-    *p_msg = msg;
-    return 0;
-}
-
-static void
-_Py_excinfo_Clear(_Py_excinfo *info)
-{
-    if (info->type != NULL) {
-        PyMem_RawFree((void *)info->type);
-    }
-    if (info->msg != NULL) {
-        PyMem_RawFree((void *)info->msg);
-    }
-    *info = (_Py_excinfo){ NULL };
-}
-
-static const char *
-_Py_excinfo_InitFromException(_Py_excinfo *info, PyObject *exc)
-{
-    assert(exc != NULL);
-
-    // Extract the exception type name.
-    const char *typename = NULL;
-    if (_exc_type_name_as_utf8(exc, &typename) < 0) {
-        assert(typename != NULL);
-        return typename;
-    }
-
-    // Extract the exception message.
-    const char *msg = NULL;
-    if (_exc_msg_as_utf8(exc, &msg) < 0) {
-        assert(msg != NULL);
-        return msg;
-    }
-
-    info->type = typename;
-    info->msg = msg;
-    return NULL;
-}
-
-static void
-_Py_excinfo_Apply(_Py_excinfo *info, PyObject *exctype)
-{
-    if (info->type != NULL) {
-        if (info->msg != NULL) {
-            PyErr_Format(exctype, "%s: %s",  info->type, info->msg);
-        }
-        else {
-            PyErr_SetString(exctype, info->type);
-        }
-    }
-    else if (info->msg != NULL) {
-        PyErr_SetString(exctype, info->msg);
-    }
-    else {
-        PyErr_SetNone(exctype);
-    }
-}
-
-
 /***************************/
 /* short-term data sharing */
 /***************************/
@@ -1051,7 +928,7 @@ _PyXI_InitExceptionInfo(_PyXI_exception_info *info,
     return failure;
 }
 
-void
+static void
 _PyXI_ApplyExceptionInfo(_PyXI_exception_info *info, PyObject *exctype)
 {
     if (exctype == NULL) {
@@ -1707,6 +1584,26 @@ _PyXI_HasCapturedException(_PyXI_session *session)
     return session->exc != NULL;
 }
 
+PyObject *
+_PyXI_ResolveCapturedException(_PyXI_session *session, PyObject *excwrapper)
+{
+    assert(!PyErr_Occurred());
+    assert(session->exc != NULL);
+    PyObject *snapshot = NULL;
+    if (session->exc->code == _PyXI_ERR_UNCAUGHT_EXCEPTION) {
+        snapshot = PyExceptionSnapshot_FromInfo(&session->exc->uncaught);
+        if (snapshot == NULL) {
+            return NULL;
+        }
+        assert(!PyErr_Occurred());
+    }
+    else {
+        _PyXI_ApplyCapturedException(session, excwrapper);
+        assert(PyErr_Occurred());
+    }
+    return snapshot;
+}
+
 int
 _PyXI_Enter(_PyXI_session *session,
             PyInterpreterState *interp, PyObject *nsupdates)
diff --git a/Python/errors.c b/Python/errors.c
index ed5eec5c261970..c55ebfdb502d61 100644
--- a/Python/errors.c
+++ b/Python/errors.c
@@ -1934,3 +1934,178 @@ PyErr_ProgramTextObject(PyObject *filename, int lineno)
 {
     return _PyErr_ProgramDecodedTextObject(filename, lineno, NULL);
 }
+
+
+/***********************/
+/* exception snapshots */
+/***********************/
+
+static const char *
+_copy_raw_string(const char *str)
+{
+    char *copied = PyMem_RawMalloc(strlen(str)+1);
+    if (copied == NULL) {
+        return NULL;
+    }
+    strcpy(copied, str);
+    return copied;
+}
+
+static int
+_exc_type_name_as_utf8(PyObject *exc, const char **p_typename)
+{
+    // XXX Use PyObject_GetAttrString(Py_TYPE(exc), '__name__')?
+    PyObject *nameobj = PyUnicode_FromString(Py_TYPE(exc)->tp_name);
+    if (nameobj == NULL) {
+        assert(PyErr_Occurred());
+        *p_typename = "unable to format exception type name";
+        return -1;
+    }
+    const char *name = PyUnicode_AsUTF8(nameobj);
+    if (name == NULL) {
+        assert(PyErr_Occurred());
+        Py_DECREF(nameobj);
+        *p_typename = "unable to encode exception type name";
+        return -1;
+    }
+    name = _copy_raw_string(name);
+    Py_DECREF(nameobj);
+    if (name == NULL) {
+        *p_typename = "out of memory copying exception type name";
+        return -1;
+    }
+    *p_typename = name;
+    return 0;
+}
+
+static int
+_exc_msg_as_utf8(PyObject *exc, const char **p_msg)
+{
+    PyObject *msgobj = PyObject_Str(exc);
+    if (msgobj == NULL) {
+        assert(PyErr_Occurred());
+        *p_msg = "unable to format exception message";
+        return -1;
+    }
+    const char *msg = PyUnicode_AsUTF8(msgobj);
+    if (msg == NULL) {
+        assert(PyErr_Occurred());
+        Py_DECREF(msgobj);
+        *p_msg = "unable to encode exception message";
+        return -1;
+    }
+    msg = _copy_raw_string(msg);
+    Py_DECREF(msgobj);
+    if (msg == NULL) {
+        assert(PyErr_ExceptionMatches(PyExc_MemoryError));
+        *p_msg = "out of memory copying exception message";
+        return -1;
+    }
+    *p_msg = msg;
+    return 0;
+}
+
+void
+_Py_excinfo_Clear(_Py_excinfo *info)
+{
+    if (info->type != NULL) {
+        PyMem_RawFree((void *)info->type);
+    }
+    if (info->msg != NULL) {
+        PyMem_RawFree((void *)info->msg);
+    }
+    *info = (_Py_excinfo){ NULL };
+}
+
+int
+_Py_excinfo_Copy(_Py_excinfo *dest, _Py_excinfo *src)
+{
+    // XXX Clear dest first?
+
+    if (src->type == NULL) {
+        dest->type = NULL;
+    }
+    else {
+        dest->type = _copy_raw_string(src->type);
+        if (dest->type == NULL) {
+            return -1;
+        }
+    }
+
+    if (src->msg == NULL) {
+        dest->msg = NULL;
+    }
+    else {
+        dest->msg = _copy_raw_string(src->msg);
+        if (dest->msg == NULL) {
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+const char *
+_Py_excinfo_InitFromException(_Py_excinfo *info, PyObject *exc)
+{
+    assert(exc != NULL);
+
+    // Extract the exception type name.
+    const char *typename = NULL;
+    if (_exc_type_name_as_utf8(exc, &typename) < 0) {
+        assert(typename != NULL);
+        return typename;
+    }
+
+    // Extract the exception message.
+    const char *msg = NULL;
+    if (_exc_msg_as_utf8(exc, &msg) < 0) {
+        assert(msg != NULL);
+        return msg;
+    }
+
+    info->type = typename;
+    info->msg = msg;
+    return NULL;
+}
+
+void
+_Py_excinfo_Apply(_Py_excinfo *info, PyObject *exctype)
+{
+    if (info->type != NULL) {
+        if (info->msg != NULL) {
+            PyErr_Format(exctype, "%s: %s",  info->type, info->msg);
+        }
+        else {
+            PyErr_SetString(exctype, info->type);
+        }
+    }
+    else if (info->msg != NULL) {
+        PyErr_SetString(exctype, info->msg);
+    }
+    else {
+        PyErr_SetNone(exctype);
+    }
+}
+
+const char *
+_Py_excinfo_AsUTF8(_Py_excinfo *info, char *buf, size_t bufsize)
+{
+    // XXX Dynamically allocate if no buf provided?
+    assert(buf != NULL);
+    if (info->type != NULL) {
+        if (info->msg != NULL) {
+            snprintf(buf, bufsize, "%s: %s",  info->type, info->msg);
+            return buf;
+        }
+        else {
+            return info->type;
+        }
+    }
+    else if (info->msg != NULL) {
+        return info->msg;
+    }
+    else {
+        return NULL;
+    }
+}
diff --git a/Python/pylifecycle.c b/Python/pylifecycle.c
index ac8d5208322882..f5ebd7e5572533 100644
--- a/Python/pylifecycle.c
+++ b/Python/pylifecycle.c
@@ -1742,7 +1742,7 @@ finalize_interp_types(PyInterpreterState *interp)
 {
     _PyUnicode_FiniTypes(interp);
     _PySys_FiniTypes(interp);
-    _PyExc_Fini(interp);
+    _PyExc_FiniTypes(interp);
     _PyAsyncGen_Fini(interp);
     _PyContext_Fini(interp);
     _PyFloat_FiniType(interp);
@@ -1779,7 +1779,7 @@ finalize_interp_clear(PyThreadState *tstate)
     int is_main_interp = _Py_IsMainInterpreter(tstate->interp);
 
     _PyXI_Fini(tstate->interp);
-    _PyExc_ClearExceptionGroupType(tstate->interp);
+    _PyExc_FiniHeapObjects(tstate->interp);
     _Py_clear_generic_types(tstate->interp);
 
     /* Clear interpreter state and all thread states */
@@ -1799,6 +1799,7 @@ finalize_interp_clear(PyThreadState *tstate)
         _PyPerfTrampoline_Fini();
     }
 
+    _PyExc_Fini(tstate->interp);
     finalize_interp_types(tstate->interp);
 }