celery: Task, AsyncResult and Signature return types can be covariant

celery-stubs/app/task.pyi

@@ -22,7 +22,7 @@ from celery.worker.request import _DeliveryInfo
 from typing_extensions import ParamSpec
 
 _P = ParamSpec("_P")
-_R = TypeVar("_R")
+_R = TypeVar("_R", covariant=True)
 _SigR = TypeVar("_SigR")
 
 class Context:

celery-stubs/canvas.pyi

@@ -15,7 +15,7 @@ from celery.result import EagerResult
 from celery.utils import abstract
 
 _F = TypeVar("_F", bound=Callable[..., Any])
-_R = TypeVar("_R")
+_R = TypeVar("_R", covariant=True)
 
 class Signature(dict[str, Any], Generic[_R]):
     @classmethod

celery-stubs/result.pyi

@@ -25,7 +25,7 @@ class ResultBase:
 
 _State = Literal["PENDING", "STARTED", "RETRY", "FAILURE", "SUCCESS"]
 
-_R = TypeVar("_R")
+_R = TypeVar("_R", covariant=True)
 
 class AsyncResult(ResultBase, Generic[_R]):
     app: Celery

tests/test_celery.py

@@ -207,6 +207,22 @@ def add(x: int, y: int) -> None:
     add.delay(1, 2)
 
 
+def test_covariant_assignment() -> None:
+    @app.task()
+    def foo() -> None:
+        pass
+
+    sig: Signature[None] = foo.s()
+    result: AsyncResult[None] = sig.apply_async()
+
+    def test_covariance(
+        t: Task[[], object], s: Signature[object], r: AsyncResult[object]
+    ) -> None:
+        pass
+
+    test_covariance(foo, sig, result)
+
+
 def test_link() -> None:
     @app.task()
     def foo() -> None: