youknowone · Oct 12, 2026 · Oct 11, 2026 · Oct 11, 2026 · Oct 11, 2026 · Oct 11, 2026
diff --git a/crates/vm/src/builtins/genericalias.rs b/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,24 @@ impl PyGenericAlias {
         Self::new(origin, args, false, vm)
     }
 
+    fn parameters(&self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
+        let cached = self.parameters.lock().clone();
+        if let Some(parameters) = cached {
+            return Ok(parameters);
+        }
+        // Attribute lookup can re-enter this alias. Compute without the lock,
+        // then replace even a cache populated by the re-entrant lookup.
+        let parameters = make_parameters(&self.args, vm);
+        let previous = core::mem::replace(
+            &mut *self.parameters.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
+    }
+
     fn repr(&self, vm: &VirtualMachine) -> PyResult<String> {
         fn repr_item(obj: &PyObject, vm: &VirtualMachine) -> PyResult<String> {
             if obj.is(&vm.ctx.ellipsis) {
@@ -206,7 +223,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 +254,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]
@@ -317,6 +335,12 @@ pub(crate) fn make_parameters(args: &Py<PyTuple>, vm: &VirtualMachine) -> PyResu
 }
 
 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 +350,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 +375,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 +585,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)?

diff --git a/extra_tests/snippets/stdlib_genericalias.py b/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