Thanks for the clear report — this is a typing variance mismatch, not a runtime bug. What’s happening - In Prefect 3.x, the

@flow

decorator parameter is typed as:

task_runner: Optional[TaskRunner[PrefectFuture[Any]]] = None

(confirmed by the current signature of

flow.__call__

). -

RayTaskRunner

is generic over its own future type (e.g.,

PrefectRayFuture[T]

). - The

TaskRunner

generic type parameter

F

is currently invariant. So a

TaskRunner[PrefectRayFuture[Any]]

is not assignable to

TaskRunner[PrefectFuture[Any]]

, even though

PrefectRayFuture

is a subtype of

PrefectFuture

. Would making

F

covariant fix it? Yes — and it’s the correct/safe change. A quick scan shows

F

is only used in return positions in

TaskRunner

(e.g., `submit()`/`map()`), so covariance is semantically appropriate. Making

F

covariant allows

RayTaskRunner[...]

to be accepted wherever a

TaskRunner[PrefectFuture[Any]]

is expected. Upstream fix (recommended) - In

src/prefect/task_runners.py

, change:

F = TypeVar("F", bound=PrefectFuture[Any], default=PrefectConcurrentFuture[Any])

to:

F = TypeVar("F", bound=PrefectFuture[Any], default=PrefectConcurrentFuture[Any], covariant=True)

This resolves the pyright/mypy error for

@flow(task_runner=RayTaskRunner())

without API changes and retains type safety. Relevant source - TaskRunner and flow decorator: - prefect/src/prefect/task_runners.py - prefect/src/prefect/flows.py - Futures: - prefect/src/prefect/futures.py - Ray runner: - prefect-ray/src/prefect_ray/task_runners.py Downstream workarounds you can use today Until the upstream change lands, here are minimally invasive options: - Cast when passing the task runner

from typing import Any, cast
from prefect import flow
from prefect.futures import PrefectFuture
from prefect.task_runners import TaskRunner
from prefect_ray.task_runners import RayTaskRunner

ray_runner = cast(TaskRunner[PrefectFuture[Any]], RayTaskRunner())

@flow(task_runner=ray_runner)
def my_flow():
    ...

- Ignore the specific assignment error (pyright/mypy)

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(task_runner=RayTaskRunner())  # type: ignore[arg-type]
def my_flow():
    ...

- Annotate a variable and ignore on the assignment

from typing import Any
from prefect.task_runners import TaskRunner
from prefect.futures import PrefectFuture
from prefect_ray.task_runners import RayTaskRunner

runner: TaskRunner[PrefectFuture[Any]]
runner = RayTaskRunner()  # type: ignore[assignment]

@flow(task_runner=runner)
def my_flow():
    ...

Notes on alternatives - Widening the

flow

parameter to

TaskRunner[Any]

would silence the error but loses useful type information; covariance on

F

is the better fix. - A Protocol-based approach is possible but unnecessary if we make

F

covariant. If you’d like, I can open an issue/PR to make

F

covariant in

TaskRunner

upstream. Also, which type checker (pyright or mypy) and versions are you using, and what versions of Prefect and prefect-ray are installed? That’ll help us validate the fix against your environment.

Done! I created a GitHub discussion to propose the upstream typing fix and included reproduction, rationale, and workarounds: - Discussion: Typing: Make TaskRunner generic parameter covariant so @flow(task_runner=RayTaskRunner()) type-checks In the meantime, you can use one of these workarounds: - Cast when passing the runner

from typing import Any, cast
from prefect import flow
from prefect.futures import PrefectFuture
from prefect.task_runners import TaskRunner
from prefect_ray.task_runners import RayTaskRunner

runner = cast(TaskRunner[PrefectFuture[Any]], RayTaskRunner())

@flow(task_runner=runner)
def my_flow():
    ...

- Or add a narrow ignore

from prefect import flow
from prefect_ray.task_runners import RayTaskRunner

@flow(task_runner=RayTaskRunner())  # type: ignore[arg-type]
def my_flow():
    ...

If you can share your type checker (pyright or mypy) and versions of Prefect/prefect-ray you’re on, I’ll add that to the thread for context.