Reject stale GenericAlias parameters on Python 3.15 (#9069)

* Reject GenericAlias parameters added after caching

Validate stale cached parameters and retain the Python regression without obsolete reference prerequisites.

Assisted-by: OpenAI Codex:model identifier unavailable

* Compute GenericAlias parameters lazily for Python 3.15

Populate the parameter cache on first lookup or subscription. Preserve cached identity and outer lookup replacement, propagate descriptor failures, and release displaced parameters outside the cache lock. Add Python regressions for timing, retries, reentrancy, recursion and concurrent callbacks.

Assisted-by: OpenAI Codex:model identifier unavailable

* Check GenericAlias recursion at lazy parameter materialization

Align the existing cyclic and deeply nested fixtures with certified CPython 3.15. Keep RecursionError assertions, cover both parameter lookup and subscription on both interpreters, and retain the separate deep-hash regression.

Assisted-by: OpenAI Codex:model identifier unavailable

* Keep finite GenericAlias stack stress native-specific

Preserve shared cyclic recursion checks and all RustPython stress assertions. Finite acyclic depth is not a portable CPython C-stack limit across optimized builds.

Assisted-by: OpenAI Codex:model identifier unavailable

* Materialize union parameters lazily like GenericAlias

Share the reentrant-safe parameter cache between aliases and unions. Preserve parameter identity without applying union argument equality or flattening. Cover nested alias timing, descriptor retry, reentrant replacement, empty caches and concurrent callbacks in Python regressions checked against final CPython 3.15.

Assisted-by: OpenAI Codex:model identifier unavailable

* Consolidate union regressions in the typing snippet

Preserve all existing regression bodies and coverage while following the requested snippet organization.

Assisted-by: OpenAI Codex:model identifier unavailable

crates/vm/src/builtins/genericalias.rs

@@ -1,5 +1,5 @@
 // spell-checker:ignore iparam gaiterobject
-use crate::common::lock::LazyLock;
+use crate::common::lock::{LazyLock, PyMutex};
 
 use super::type_;
 use crate::{
@@ -40,7 +40,7 @@ pub struct PyGenericAlias {
     origin: PyObjectRef,
     #[pymember(name = "__args__")]
     args: PyTupleRef,
-    parameters: PyTupleRef,
+    parameters: PyMutex<Option<PyTupleRef>>,
     #[pymember(name = "__unpacked__")]
     starred: bool, // for __unpacked__ attribute
 }
@@ -105,13 +105,12 @@ impl PyGenericAlias {
         origin: impl Into<PyObjectRef>,
         args: PyTupleRef,
         starred: bool,
-        vm: &VirtualMachine,
+        _vm: &VirtualMachine,
     ) -> PyResult<Self> {
-        let parameters = make_parameters(&args, vm)?;
         Ok(Self {
             origin: origin.into(),
             args,
-            parameters,
+            parameters: PyMutex::new(None),
             starred,
         })
     }
@@ -130,6 +129,10 @@ impl PyGenericAlias {
         Self::new(origin, args, false, vm)
     }
 
+    fn parameters(&self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
+        cached_parameters(&self.parameters, &self.args, vm)
+    }
+
     fn repr(&self, vm: &VirtualMachine) -> PyResult<String> {
         fn repr_item(obj: &PyObject, vm: &VirtualMachine) -> PyResult<String> {
             if obj.is(&vm.ctx.ellipsis) {
@@ -206,7 +209,8 @@ impl PyGenericAlias {
     }
 
     fn __getitem__(zelf: &Py<Self>, needle: PyObjectRef, vm: &VirtualMachine) -> PyResult {
-        let new_args = subs_parameters(zelf.as_object(), &zelf.args, &zelf.parameters, needle, vm)?;
+        let parameters = zelf.parameters(vm)?;
+        let new_args = subs_parameters(zelf.as_object(), &zelf.args, &parameters, needle, vm)?;
 
         Ok(Self::new(zelf.origin.clone(), new_args, false, vm)?.into_pyobject(vm))
     }
@@ -236,8 +240,8 @@ impl PyGenericAlias {
 )]
 impl Py<PyGenericAlias> {
     #[pygetset]
-    fn __parameters__(&self) -> PyObjectRef {
-        self.parameters.clone().into()
+    fn __parameters__(&self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
+        self.parameters(vm)
     }
 
     #[pygetset]
@@ -312,11 +316,36 @@ impl Py<PyGenericAlias> {
     }
 }
 
+pub(crate) fn cached_parameters(
+    cache: &PyMutex<Option<PyTupleRef>>,
+    args: &Py<PyTuple>,
+    vm: &VirtualMachine,
+) -> PyResult<PyTupleRef> {
+    let cached = cache.lock().clone();
+    if let Some(parameters) = cached {
+        return Ok(parameters);
+    }
+    // Attribute lookup can re-enter this alias or union. Compute without the
+    // lock, then replace even a cache populated by the re-entrant lookup.
+    let parameters = make_parameters(args, vm);
+    let previous = core::mem::replace(&mut *cache.lock(), parameters.as_ref().ok().cloned());
+    // A failed outer lookup clears the cache populated by a nested lookup too.
+    // Cached parameters can own objects with Python finalizers.
+    drop(previous);
+    parameters
+}
+
 pub(crate) fn make_parameters(args: &Py<PyTuple>, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
     make_parameters_from_slice(args.as_slice(), vm)
 }
 
 fn make_parameters_from_slice(args: &[PyObjectRef], vm: &VirtualMachine) -> PyResult<PyTupleRef> {
+    vm.with_recursion(" in __parameter__ calculation", || {
+        make_parameters_inner(args, vm)
+    })
+}
+
+fn make_parameters_inner(args: &[PyObjectRef], vm: &VirtualMachine) -> PyResult<PyTupleRef> {
     let mut parameters: Vec<PyObjectRef> = Vec::with_capacity(args.len());
 
     for arg in args {
@@ -326,16 +355,21 @@ fn make_parameters_from_slice(args: &[PyObjectRef], vm: &VirtualMachine) -> PyRe
         }
 
         // Check for __typing_subst__ attribute
-        if arg.get_attr(identifier!(vm, __typing_subst__), vm).is_ok() {
+        if vm
+            .get_attribute_opt(arg, identifier!(vm, __typing_subst__))?
+            .is_some()
+        {
             if tuple_index(&parameters, arg).is_none() {
                 parameters.push(arg.clone());
             }
-        } else if let Ok(subparams) = arg.get_attr(identifier!(vm, __parameters__), vm)
-            && let Ok(sub_params) = subparams.try_to_ref::<PyTuple>(vm)
+        } else if let Some(subparams) =
+            vm.get_attribute_opt(arg, identifier!(vm, __parameters__))?
         {
-            for sub_param in sub_params {
-                if tuple_index(&parameters, sub_param).is_none() {
-                    parameters.push(sub_param.clone());
+            if let Some(sub_params) = subparams.downcast_ref::<PyTuple>() {
+                for sub_param in sub_params {
+                    if tuple_index(&parameters, sub_param).is_none() {
+                        parameters.push(sub_param.clone());
+                    }
                 }
             }
         } else if arg.try_to_ref::<PyTuple>(vm).is_ok() || arg.try_to_ref::<PyList>(vm).is_ok() {
@@ -346,9 +380,7 @@ fn make_parameters_from_slice(args: &[PyObjectRef], vm: &VirtualMachine) -> PyRe
                 let list = arg.downcast_ref::<PyList>().unwrap();
                 list.borrow_vec().to_vec()
             };
-            let sub = vm.with_recursion(" while computing __parameters__", || {
-                make_parameters_from_slice(&items, vm)
-            })?;
+            let sub = make_parameters_from_slice(&items, vm)?;
             for sub_param in sub.as_slice() {
                 if tuple_index(&parameters, sub_param).is_none() {
                     parameters.push(sub_param.clone());
@@ -558,7 +590,10 @@ pub(crate) fn subs_parameters(
             if let Some(iparam) = tuple_index(parameters.as_slice(), arg) {
                 subst.call((arg_items[iparam].clone(),), vm)?
             } else {
-                subs_tvars(arg.clone(), parameters, &arg_items, vm)?
+                return Err(vm.new_type_error(format!(
+                    "argument {} with __typing_subst__ was not found in __parameters__",
+                    arg.repr(vm)?
+                )));
             }
         } else {
             subs_tvars(arg.clone(), parameters, &arg_items, vm)?

crates/vm/src/builtins/union.rs

@@ -1,5 +1,5 @@
 use super::{genericalias, type_};
-use crate::common::lock::LazyLock;
+use crate::common::lock::{LazyLock, PyMutex};
 use crate::{
     AsObject, Context, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine,
     atomic_func,
@@ -24,7 +24,7 @@ pub struct PyUnion {
     hashable_args: Option<PyRef<PyFrozenSet>>,
     /// Tuple of initially unhashable args, or None if all args were hashable
     unhashable_args: Option<PyTupleRef>,
-    parameters: PyTupleRef,
+    parameters: PyMutex<Option<PyTupleRef>>,
 }
 
 impl fmt::Debug for PyUnion {
@@ -42,14 +42,13 @@ impl PyPayload for PyUnion {
 
 impl PyUnion {
     /// Create a new union from dedup result (internal use)
-    fn from_components(result: UnionComponents, vm: &VirtualMachine) -> PyResult<Self> {
-        let parameters = make_parameters(&result.args, vm)?;
-        Ok(Self {
+    fn from_components(result: UnionComponents) -> Self {
+        Self {
             args: result.args,
             hashable_args: result.hashable_args,
             unhashable_args: result.unhashable_args,
-            parameters,
-        })
+            parameters: PyMutex::new(None),
+        }
     }
 
     /// Direct access to args field (_Py_union_args)
@@ -127,8 +126,8 @@ impl Py<PyUnion> {
     }
 
     #[pygetset]
-    fn __parameters__(&self) -> PyObjectRef {
-        self.parameters.clone().into()
+    fn __parameters__(&self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
+        genericalias::cached_parameters(&self.parameters, &self.args, vm)
     }
 
     #[pymethod]
@@ -239,12 +238,6 @@ pub(crate) fn or_op(zelf: PyObjectRef, other: PyObjectRef, vm: &VirtualMachine)
     make_union(&tuple, vm)
 }
 
-fn make_parameters(args: &Py<PyTuple>, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
-    let parameters = genericalias::make_parameters(args, vm)?;
-    let result = dedup_and_flatten_args(&parameters, vm)?;
-    Ok(result.args)
-}
-
 fn flatten_args(args: &Py<PyTuple>, vm: &VirtualMachine) -> PyTupleRef {
     let mut total_args = 0;
     for arg in args {
@@ -372,19 +365,15 @@ pub fn make_union(args: &Py<PyTuple>, vm: &VirtualMachine) -> PyResult {
     let result = dedup_and_flatten_args(args, vm)?;
     Ok(match result.args.as_slice().len() {
         1 => result.args.as_slice()[0].to_owned(),
-        _ => PyUnion::from_components(result, vm)?.to_pyobject(vm),
+        _ => PyUnion::from_components(result).to_pyobject(vm),
     })
 }
 
 impl PyUnion {
     fn getitem(zelf: &Py<Self>, needle: PyObjectRef, vm: &VirtualMachine) -> PyResult {
-        let new_args = genericalias::subs_parameters(
-            zelf.as_object(),
-            &zelf.args,
-            &zelf.parameters,
-            needle,
-            vm,
-        )?;
+        let parameters = genericalias::cached_parameters(&zelf.parameters, &zelf.args, vm)?;
+        let new_args =
+            genericalias::subs_parameters(zelf.as_object(), &zelf.args, &parameters, needle, vm)?;
 
         Ok(if new_args.as_slice().is_empty() {
             make_union(&new_args, vm)?

extra_tests/snippets/stdlib_genericalias.py

@@ -0,0 +1,226 @@
+from types import GenericAlias
+
+from testutils import assert_raises
+
+
+class Parameter:
+    def __typing_subst__(self, value):
+        return value
+
+
+parameter = Parameter()
+
+
+# Construction does not inspect arguments. Both entry points populate the cache
+# using the attributes present at the first lookup, rather than construction.
+class LateParameter:
+    __parameters__ = (parameter,)
+
+    def __repr__(self):
+        return "late_parameter"
+
+
+for materialize in (False, True):
+    late = LateParameter()
+    alias = GenericAlias(list, late)
+    if materialize:
+        cached = alias.__parameters__
+        assert cached == (parameter,)
+    late.__typing_subst__ = lambda value: value
+    if materialize:
+        with assert_raises(TypeError) as caught:
+            alias[int]
+        assert str(caught.exception) == (
+            "argument late_parameter with __typing_subst__ was not found in __parameters__"
+        )
+        assert alias.__parameters__ is cached
+    else:
+        assert alias[int] == list[int]
+        cached = alias.__parameters__
+        assert cached == (late,)
+        assert alias.__parameters__ is cached
+
+late = LateParameter()
+alias = GenericAlias(list, late)
+late.__typing_subst__ = lambda value: value
+assert alias.__parameters__ == (late,)
+assert alias[int] == list[int]
+
+# Empty tuples are cached too, even if parameters appear later.
+empty = LateParameter()
+empty.__parameters__ = ()
+alias = GenericAlias(list, empty)
+cached = alias.__parameters__
+empty.__typing_subst__ = lambda value: value
+assert alias.__parameters__ is cached
+assert cached == ()
+with assert_raises(TypeError):
+    alias[int]
+
+
+# Lookup errors happen lazily and leave the cache uninitialized for a retry.
+class FailingParameter:
+    def __init__(self, failing_attribute):
+        self.failing_attribute = failing_attribute
+        self.fail = True
+        self.lookups = []
+
+    def __getattribute__(self, name):
+        if name in ("__typing_subst__", "__parameters__"):
+            self.lookups.append(name)
+            if name == self.failing_attribute and self.fail:
+                raise RuntimeError("parameter lookup failed")
+        return object.__getattribute__(self, name)
+
+    __parameters__ = (parameter,)
+
+
+for attribute in ("__typing_subst__", "__parameters__"):
+    for subscribe in (False, True):
+        failing = FailingParameter(attribute)
+        alias = GenericAlias(list, failing)
+        assert failing.lookups == []
+        with assert_raises(RuntimeError) as caught:
+            if subscribe:
+                alias[int]
+            else:
+                alias.__parameters__
+        assert str(caught.exception) == "parameter lookup failed"
+        failing.fail = False
+        cached = alias.__parameters__
+        assert cached == (parameter,)
+        count = len(failing.lookups)
+        assert alias.__parameters__ is cached
+        assert len(failing.lookups) == count
+
+
+# A nested lookup may populate the cache, but the outer calculation replaces it.
+inner_parameter = Parameter()
+outer_parameter = Parameter()
+
+
+class ReentrantParameter:
+    entered = False
+
+    @property
+    def __parameters__(self):
+        if not self.entered:
+            self.entered = True
+            self.inner = self.alias.__parameters__
+            return (outer_parameter,)
+        return (inner_parameter,)
+
+
+reentrant = ReentrantParameter()
+reentrant.alias = GenericAlias(list, reentrant)
+cached = reentrant.alias.__parameters__
+assert reentrant.inner == (inner_parameter,)
+assert cached == (outer_parameter,)
+assert cached is reentrant.alias.__parameters__
+assert cached is not reentrant.inner
+
+
+# A cyclic parameter lookup raises rather than deadlocking or overflowing Rust.
+class RecursiveParameter:
+    recurse = True
+
+    @property
+    def __parameters__(self):
+        if self.recurse:
+            return self.alias.__parameters__
+        return (parameter,)
+
+
+recursive = RecursiveParameter()
+recursive.alias = GenericAlias(list, recursive)
+with assert_raises(RecursionError):
+    recursive.alias.__parameters__
+recursive.recurse = False
+assert recursive.alias.__parameters__ == (parameter,)
+
+# Substitution and unpacking construct aliases with their own lazy caches.
+failing = FailingParameter("__parameters__")
+result = GenericAlias(list, parameter)[failing]
+assert failing.lookups == []
+with assert_raises(RuntimeError):
+    result.__parameters__
+
+failing = FailingParameter("__parameters__")
+original = GenericAlias(tuple, failing)
+starred = next(iter(original))
+assert failing.lookups == []
+with assert_raises(RuntimeError):
+    starred.__parameters__
+failing.fail = False
+assert starred.__parameters__ == (parameter,)
+failing.__parameters__ = ()
+assert original.__parameters__ == ()
+assert starred.__parameters__ == (parameter,)
+
+# Concurrent descriptor callbacks can overlap without holding the cache lock.
+import threading
+
+barrier = threading.Barrier(2)
+
+
+class ConcurrentParameter:
+    @property
+    def __parameters__(self):
+        barrier.wait(timeout=10)
+        return (parameter,)
+
+
+concurrent = GenericAlias(list, ConcurrentParameter())
+results = []
+errors = []
+
+
+def read_parameters():
+    try:
+        results.append(concurrent.__parameters__)
+    except BaseException as error:
+        errors.append(error)
+
+
+threads = [threading.Thread(target=read_parameters) for _ in range(2)]
+for thread in threads:
+    thread.start()
+for thread in threads:
+    thread.join(timeout=15)
+assert all(not thread.is_alive() for thread in threads)
+assert errors == []
+assert results == [(parameter,), (parameter,)]
+cached = concurrent.__parameters__
+assert any(cached is result for result in results)
+assert concurrent.__parameters__ is cached
+
+
+# Failure in the outer calculation clears an inner successful cache as well.
+class ReentrantFailure:
+    entered = False
+    fail = True
+
+    @property
+    def __parameters__(self):
+        if not self.entered:
+            self.entered = True
+            self.inner = self.alias.__parameters__
+            if self.fail:
+                raise RuntimeError("outer lookup failed")
+        return (inner_parameter,) if self.fail else (outer_parameter,)
+
+
+for subscribe in (False, True):
+    failure = ReentrantFailure()
+    failure.alias = GenericAlias(list, failure)
+    with assert_raises(RuntimeError) as caught:
+        if subscribe:
+            failure.alias[int]
+        else:
+            failure.alias.__parameters__
+    assert str(caught.exception) == "outer lookup failed"
+    assert failure.inner == (inner_parameter,)
+    failure.fail = False
+    cached = failure.alias.__parameters__
+    assert cached == (outer_parameter,)
+    assert failure.alias.__parameters__ is cached

extra_tests/snippets/stdlib_types.py

@@ -43,21 +43,27 @@ def _run_missing_type_params_regression():
 
 _run_missing_type_params_regression()
 
-if sys.implementation.name == "rustpython":
-    # __parameters__ is computed when the alias is built, and the walk descends
-    # into every list and tuple argument, so a self-referential or deeply
-    # nested argument must be caught. CPython, which also runs this snippet,
-    # does not walk into plain lists at all.
-    self_referential = []
-    self_referential.append(self_referential)
-    with assert_raises(RecursionError):
-        list[self_referential]
+# GenericAlias construction is lazy. Python 3.15 walks nested lists and tuples
+# when parameters are first needed, including during subscription.
+self_referential = []
+self_referential.append(self_referential)
+alias = list[self_referential]
+with assert_raises(RecursionError):
+    alias.__parameters__
+with assert_raises(RecursionError):
+    alias[int]
 
+if sys.implementation.name == "rustpython":
+    # Finite-depth stack exhaustion depends on the interpreter build and OS stack.
+    # Keep this native stress case separate from the shared cyclic contract above.
     nested = [0]
     for _ in range(100_000):
         nested = [nested]
+    alias = list[nested]
     with assert_raises(RecursionError):
-        list[nested]
+        alias.__parameters__
+    with assert_raises(RecursionError):
+        alias[int]
 
     # hashing an alias walks the same shape. Release inlining of the hash
     # slot needs more depth than debug to trip the native stack guard.

extra_tests/snippets/stdlib_typing.py

@@ -89,3 +89,217 @@ for alias_value in (int, AliasModuleValue()):
     )
     assert alias_namespace["alias"].__module__ is None
 assert alias_module_lookups == []
+
+
+# GenericAlias rejects parameters added after its parameter tuple was cached.
+from types import GenericAlias
+
+
+class LateParameter:
+    def __repr__(self):
+        return "late_parameter"
+
+
+first = LateParameter()
+first.__typing_subst__ = lambda value: value
+late = LateParameter()
+alias = GenericAlias(dict, (first, late))
+assert alias.__parameters__ == (first,)
+late.__typing_subst__ = lambda value: value
+with assert_raises(TypeError) as caught:
+    alias[0]
+assert str(caught.exception) == (
+    "argument late_parameter with __typing_subst__ was not found in __parameters__"
+)
+
+
+def test_union_lazy_parameters():
+    import threading
+    from types import GenericAlias
+    from typing import Union
+
+    from testutils import assert_raises
+
+    class Parameter:
+        def __typing_subst__(self, value):
+            return value
+
+    parameter = Parameter()
+
+    class LateParameter:
+        __parameters__ = (parameter,)
+
+        def __repr__(self):
+            return "late_parameter"
+
+    # Building a union must not freeze the nested alias's parameter cache.
+    for materialize in (False, True):
+        for subscribe_first in (False, True):
+            late = LateParameter()
+            alias = GenericAlias(list, late)
+            union = alias | int
+            if materialize:
+                cached = union.__parameters__
+                assert cached == (parameter,)
+            late.__typing_subst__ = lambda value: value
+            if materialize:
+                with assert_raises(TypeError) as caught:
+                    union[str]
+                assert str(caught.exception) == (
+                    "argument late_parameter with __typing_subst__ was not found in __parameters__"
+                )
+                assert union.__parameters__ is cached
+            else:
+                if not subscribe_first:
+                    assert union.__parameters__ == (late,)
+                assert union[str] == (list[str] | int)
+                assert alias[str] == list[str]
+                cached = union.__parameters__
+                assert cached == (late,)
+                assert union.__parameters__ is cached
+
+    # typing.Union and flattening an existing union preserve the same lazy boundary.
+    for make_union in (
+        lambda alias: Union[alias, int],
+        lambda alias: (alias | bytes) | int,
+    ):
+        late = LateParameter()
+        alias = GenericAlias(list, late)
+        union = make_union(alias)
+        late.__typing_subst__ = lambda value: value
+        assert union[str] == make_union(list[str])
+        assert alias[str] == list[str]
+
+    # Nested descriptor failures occur on first use and do not poison either cache.
+    class FailingParameter:
+        fail = True
+        lookups = 0
+
+        @property
+        def __parameters__(self):
+            self.lookups += 1
+            if self.fail:
+                raise RuntimeError("lazy union failure")
+            return (parameter,)
+
+    for subscribe in (False, True):
+        failing = FailingParameter()
+        alias = GenericAlias(list, failing)
+        union = alias | int
+        assert failing.lookups == 0
+        with assert_raises(RuntimeError) as caught:
+            if subscribe:
+                union[str]
+            else:
+                union.__parameters__
+        assert str(caught.exception) == "lazy union failure"
+        failing.fail = False
+        cached = union.__parameters__
+        assert cached == (parameter,)
+        count = failing.lookups
+        assert union.__parameters__ is cached
+        assert failing.lookups == count
+
+    # Parameter collection uses identity, not union argument hash/equality rules.
+    parameter_comparisons = []
+
+    class EqualParameter(Parameter):
+        def __hash__(self):
+            parameter_comparisons.append("hash")
+            return 0
+
+        def __eq__(self, other):
+            parameter_comparisons.append("eq")
+            return True
+
+    first = EqualParameter()
+    second = EqualParameter()
+    union = GenericAlias(tuple, (first, second)) | int
+    parameter_comparisons.clear()
+    cached = union.__parameters__
+    assert len(cached) == 2
+    assert cached[0] is first and cached[1] is second
+    assert parameter_comparisons == []
+    assert union.__parameters__ is cached
+
+    # Empty caches are also stable when a nested alias's descriptor changes.
+    state = {"parameters": ()}
+
+    class ChangingAlias(GenericAlias):
+        @property
+        def __parameters__(self):
+            return state["parameters"]
+
+    union = ChangingAlias(list, int) | str
+    cached = union.__parameters__
+    assert cached == ()
+    state["parameters"] = (parameter,)
+    assert union.__parameters__ is cached
+    with assert_raises(TypeError):
+        union[bytes]
+
+    # Reentrant callbacks may populate the cache. The outer calculation replaces
+    # that cache on success and clears it on failure, allowing another attempt.
+    inner_parameter = Parameter()
+    outer_parameter = Parameter()
+
+    for fail in (False, True):
+        state = {"entered": False, "parameters": (inner_parameter,)}
+
+        class ReentrantAlias(GenericAlias):
+            @property
+            def __parameters__(self):
+                if not state["entered"]:
+                    state["entered"] = True
+                    state["inner"] = state["union"].__parameters__
+                    if fail:
+                        raise RuntimeError("outer union failure")
+                    return (outer_parameter,)
+                return state["parameters"]
+
+        state["union"] = ReentrantAlias(list, int) | str
+        union = state["union"]
+        if fail:
+            with assert_raises(RuntimeError) as caught:
+                union.__parameters__
+            assert str(caught.exception) == "outer union failure"
+            state["parameters"] = (outer_parameter,)
+        cached = union.__parameters__
+        assert state["inner"] == (inner_parameter,)
+        assert cached == (outer_parameter,)
+        assert cached is not state["inner"]
+        assert union.__parameters__ is cached
+
+    # Concurrent callbacks must run without holding the union's parameter lock.
+    barrier = threading.Barrier(2)
+
+    class ConcurrentAlias(GenericAlias):
+        @property
+        def __parameters__(self):
+            barrier.wait(timeout=10)
+            return (parameter,)
+
+    union = ConcurrentAlias(list, int) | str
+    results = []
+    errors = []
+
+    def read_parameters():
+        try:
+            results.append(union.__parameters__)
+        except BaseException as error:
+            errors.append(error)
+
+    threads = [threading.Thread(target=read_parameters) for _ in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=15)
+    assert all(not thread.is_alive() for thread in threads)
+    assert errors == []
+    assert results == [(parameter,), (parameter,)]
+    cached = union.__parameters__
+    assert any(cached is result for result in results)
+    assert union.__parameters__ is cached
+
+
+test_union_lazy_parameters()