@@ -170,6 +170,10 @@ PyAPI_FUNC(int) _PyDict_SetItem_KnownHash_LockHeld(PyDictObject *mp, PyObject *k
 PyAPI_FUNC(int) _PyDict_GetItemRef_KnownHash_LockHeld(PyDictObject *op, PyObject *key, Py_hash_t hash, PyObject **result);
 PyAPI_FUNC(int) _PyDict_GetItemRef_KnownHash(PyDictObject *op, PyObject *key, Py_hash_t hash, PyObject **result);
 extern int _PyDict_GetItemRef_Unicode_LockHeld(PyDictObject *op, PyObject *key, PyObject **result);
+extern PyObject *_PyDict_GetItem_ExactUnicode(PyDictObject *mp, PyObject *key);
+extern int _PyDict_GetItemStackRef_ExactUnicode(PyDictObject *mp,
+                                                PyObject *key,
+                                                _PyStackRef *value_addr);
 PyAPI_FUNC(int) _PyObjectDict_SetItem(PyTypeObject *tp, PyObject *obj, PyObject **dictptr, PyObject *name, PyObject *value);
 
 extern int _PyDict_Pop_KnownHash(
 
@@ -5098,6 +5098,31 @@ def __new__(mcls, name, bases, ns):
         with self.assertWarnsRegex(RuntimeWarning, 'MyClass'):
             MyClass = meta('MyClass', (), {})
 
+    def test_str_subclass_namespace_keys(self):
+        # Keys of the class namespace that are str subclasses are copied
+        # to exact str in the type's __dict__, like type.__setattr__ does.
+        class Key(str):
+            def __hash__(self):
+                # Collide with the slot name assigned below.
+                return hash('__hash__')
+            def __eq__(self, other):
+                # Must not be called by type dict lookups.
+                raise AssertionError('Key.__eq__ called')
+
+        ns = {Key('other'): 1, '__repr__': lambda self: 'r', 'c': 2}
+        C = type('C', (), ns)
+        for key in C.__dict__:
+            self.assertIs(type(key), str)
+        self.assertEqual([str(k) for k in ns], list(C.__dict__)[:3])
+        self.assertEqual(repr(C()), 'r')
+        self.assertEqual(C.other, 1)
+        # Lookups and slot updates that hash-collide with 'other' do not
+        # call Key.__eq__.
+        C.__hash__ = None
+        with self.assertRaises(TypeError):
+            hash(C())
+        del C.__hash__
+
     def test_proxy_call(self):
         class FakeStr:
             __class__ = str
 
@@ -26,6 +26,30 @@ def mutate_thread():
 
 
 class TestThreadState(unittest.TestCase):
+    def test_stop_the_world(self):
+        import _testinternalcapi
+
+        _testinternalcapi.stop_the_world()
+        _testinternalcapi.stop_the_world(0)
+
+        def pause():
+            for _ in range(10):
+                _testinternalcapi.stop_the_world(0.0001)
+
+        threading_helper.run_concurrently(pause, nthreads=2)
+        for duration in (-1, -0.1, float("nan")):
+            with self.subTest(duration=duration):
+                with self.assertRaises(ValueError):
+                    _testinternalcapi.stop_the_world(duration)
+        for duration in (float("inf"), 10**100):
+            with self.subTest(duration=duration):
+                with self.assertRaises(OverflowError):
+                    _testinternalcapi.stop_the_world(duration)
+        for duration in (None, "0"):
+            with self.subTest(duration=duration):
+                with self.assertRaises(TypeError):
+                    _testinternalcapi.stop_the_world(duration)
+
     @support.requires_subprocess()
     def test_tight_stw_loop_does_not_starve_attach(self):
         script = textwrap.dedent(f"""
 
@@ -1,3 +1,4 @@
+import gc
 import threading
 import unittest
 
@@ -19,6 +20,24 @@
 class A:
     attr = 1
 
+def world_stoppers():
+    """Finite workloads that contend for the stop-the-world mutex.
+
+    Waiting for the mutex releases the type lock, so that other threads
+    can change types in the meantime.  Use a short pause rather than full collections
+    to prolong the contention without depending on the size of the heap.
+    """
+    def stopper():
+        for _ in range(100):
+            _testinternalcapi.stop_the_world()
+
+    def pauser():
+        for _ in range(100):
+            _testinternalcapi.stop_the_world(0.0001)
+
+    return [stopper, pauser]
+
+
 @threading_helper.requires_working_threading()
 class TestType(TestCase):
     def test_attr_cache(self):
@@ -325,9 +344,8 @@ def wrapper():
             reader.join()
 
     def test_setattr_many_subclasses(self):
-        # gh-155978: Updating a special method queues a slot update for every
-        # affected subclass.  Keep enough subclasses alive to require
-        # heap-allocated queue chunks in addition to the stack chunk.
+        # gh-155978: Updating a special method updates the slot of every
+        # affected subclass.
         class Base:
             pass
 
@@ -344,6 +362,11 @@ def custom_repr(self):
         self.assertTrue(all(repr(cls()) != "custom repr"
                             for cls in subclasses))
 
+        # The slot update must not keep references to the types.
+        del subclasses
+        gc.collect()
+        self.assertEqual(Base.__subclasses__(), [])
+
     def test_concurrent_setattr_deadlock(self):
         # gh-155400: two threads assigning to a special method of the same
         # class could deadlock.  One thread held the type lock and waited for
@@ -401,6 +424,151 @@ class B(A):
         with threading_helper.start_threads(threads):
             pass
 
+    def check_stop_the_world_under_type_lock(self, setter):
+        # gh-158195: Python code that runs with the type lock held and stops
+        # the world could deadlock with a thread that held the type lock
+        # pinned while it waited for the stop-the-world mutex (__bases__
+        # assignment).  The first thread released the type lock while it was
+        # blocked on the stop-the-world mutex, was handed the mutex, and then
+        # waited for the type lock while owning the mutex.
+        def basesetter():
+            class A:
+                pass
+            class B:
+                pass
+            class C(A):
+                pass
+            for _ in range(100):
+                C.__bases__ = (B,)
+                C.__bases__ = (A,)
+
+        def stopper():
+            for _ in range(100):
+                _testinternalcapi.stop_the_world()
+
+        # The stopper keeps the stop-the-world mutex contended, which makes
+        # the setters block on it.  All workers have finite workloads so
+        # that repeated invalidation cannot indefinitely delay a setter.
+        threading_helper.run_concurrently(
+            [setter, setter, basesetter, stopper])
+
+    def test_stop_the_world_in_mro_deadlock(self):
+        # type_ready() calls a metaclass mro() with the type lock held.
+        N = 50
+
+        class Meta(type):
+            def mro(cls):
+                _testinternalcapi.stop_the_world()
+                return type.mro(cls)
+
+        def setter():
+            for _ in range(N):
+                Meta("K", (), {})
+
+        self.check_stop_the_world_under_type_lock(setter)
+
+    def test_reentrant_mro_always_publishes(self):
+        # A custom mro() that updates a slot on every call must not make
+        # the __bases__ assignment retry forever.
+        class Other:
+            pass
+
+        class Meta(type):
+            def mro(cls):
+                Other.__repr__ = lambda self: "x"
+                return type.mro(cls)
+
+        class A:
+            pass
+
+        class B:
+            pass
+
+        class C(A, metaclass=Meta):
+            pass
+
+        C.__bases__ = (B,)
+        self.assertEqual(C.__mro__, (C, B, object))
+
+    def test_slot_update_skips_non_str_keys(self):
+        # Slots are computed (with the world stopped, for an update) without
+        # comparing the slot name with non-str keys of the type dicts, which
+        # could run user code.  Here, that code would also update a slot
+        # every time.
+        calls = 0
+
+        class Other:
+            pass
+
+        class Key:
+            def __hash__(self):
+                return hash("__hash__")
+
+            def __eq__(self, other):
+                nonlocal calls
+                calls += 1
+                Other.__repr__ = lambda self: "x"
+                return False
+
+        class Base:
+            pass
+
+        with self.assertWarns(RuntimeWarning):
+            Sub = type("Sub", (Base,), {Key(): None})
+        self.assertEqual(calls, 0)
+
+        Base.__hash__ = None
+        self.assertEqual(calls, 0)
+        with self.assertRaises(TypeError):
+            hash(Sub())
+        del Base.__hash__
+        hash(Sub())
+
+    def test_concurrent_bases_assignment(self):
+        # Several threads reassign __bases__ of the same class, and of its
+        # base, while others stop the world.  The type lock is released while
+        # waiting for the stop-the-world.  At the end, the MRO must agree
+        # with __bases__.
+        N = 50
+
+        class A:
+            def meth(self):
+                return "A"
+            def __repr__(self):
+                return "A"
+        class B:
+            def meth(self):
+                return "B"
+            def __repr__(self):
+                return "B"
+        class C(A):
+            pass
+        class D(C):
+            pass
+
+        def setter():
+            for _ in range(N):
+                C.__bases__ = (B,)
+                C.__bases__ = (A,)
+
+        def subsetter():
+            for _ in range(N):
+                D.__bases__ = (C, B)
+                D.__bases__ = (C,)
+
+        threading_helper.run_concurrently(
+            [setter, setter, subsetter] + world_stoppers())
+
+        self.assertEqual(C.__bases__, (A,))
+        self.assertEqual(C.__mro__, (C, A, object))
+        self.assertEqual(D.__bases__, (C,))
+        self.assertEqual(D.__mro__, (D, C, A, object))
+        self.assertEqual(D().meth(), "A")
+        self.assertEqual(repr(C()), "A")
+        self.assertEqual(repr(D()), "A")
+        self.assertIn(C, A.__subclasses__())
+        self.assertNotIn(C, B.__subclasses__())
+
     def test_class_creation_with_concurrent_setattr(self):
         # Creating a class resolves its slots from the dicts along the MRO
         # while holding the type lock, before the class is reachable from
 
@@ -0,0 +1,7 @@
+Fix a deadlock in the free-threaded build between a thread that stops the
+world while running Python code under the type lock (for example a custom
+metaclass ``mro()`` or the ``__eq__`` of a :class:`str` subclass used as a
+class ``__dict__`` key) and a thread that assigns to a special method or to
+``__bases__`` of a class.  The type lock is no longer held while waiting
+for a stop-the-world; the prepared slot, MRO and bases updates are
+validated with the world stopped and recomputed if they went stale.
@@ -38,6 +38,7 @@
 #include "pycore_pylifecycle.h"   // _PyInterpreterConfig_InitFromDict()
 #include "pycore_pystate.h"       // _PyThreadState_GET()
 #include "pycore_runtime_structs.h" // _PY_NSMALLPOSINTS
+#include "pycore_time.h"          // _PyTime_FromSecondsObject()
 #include "pycore_unicodeobject.h" // _PyUnicode_TransformDecimalAndSpaceToASCII()
 
 #include "clinic/_testinternalcapi.c.h"
@@ -209,6 +210,42 @@ get_stack_margin(PyObject *self, PyObject *Py_UNUSED(args))
     return PyLong_FromSize_t(_PyOS_STACK_MARGIN_BYTES);
 }
 
+static PyObject *
+stop_the_world(PyObject *self, PyObject *args)
+{
+    PyObject *duration_obj = NULL;
+    PyTime_t duration = 0;
+    if (!PyArg_ParseTuple(args, "|O:stop_the_world", &duration_obj)) {
+        return NULL;
+    }
+    if (duration_obj != NULL &&
+        _PyTime_FromSecondsObject(&duration, duration_obj,
+                                 _PyTime_ROUND_TIMEOUT) < 0) {
+        return NULL;
+    }
+    if (duration < 0) {
+        PyErr_SetString(PyExc_ValueError, "duration must be non-negative");
+        return NULL;
+    }
+#ifdef Py_GIL_DISABLED
+    PyInterpreterState *interp = _PyInterpreterState_GET();
+    _PyEval_StopTheWorld(interp);
+    if (duration != 0) {
+        PyEvent event = {0};
+        PyEvent_WaitTimed(&event, duration, /*detach=*/0);
+    }
+    _PyEval_StartTheWorld(interp);
+#endif
+    Py_RETURN_NONE;
+}
+
+PyDoc_STRVAR(stop_the_world_doc,
+"stop_the_world($module, duration=0, /)\n"
+"--\n"
+"\n"
+"Stop the world, wait for duration seconds, then restart it.\n"
+"Does nothing in GIL-enabled builds.");
+
 static PyObject *
 test_stop_the_world(PyObject *self, PyObject *Py_UNUSED(args))
 {
@@ -3363,6 +3400,7 @@ static PyMethodDef module_functions[] = {
     {"get_c_recursion_remaining", get_c_recursion_remaining, METH_NOARGS},
     {"get_stack_pointer", get_stack_pointer, METH_NOARGS},
     {"get_stack_margin", get_stack_margin, METH_NOARGS},
+    {"stop_the_world", stop_the_world, METH_VARARGS, stop_the_world_doc},
     {"test_stop_the_world", test_stop_the_world, METH_NOARGS},
     {"classify_stack_addresses", classify_stack_addresses, METH_VARARGS},
     {"get_jit_code_ranges", get_jit_code_ranges, METH_NOARGS},