Cannot pass core.FlowRun instance as argument to task. Raises prefect.exceptions.MissingResult

[jborman-exos]

First check

• I added a descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the Prefect documentation for this issue.
• I checked that this issue is related to Prefect and not one of its dependencies.

Bug summary

Bug: I cannot pass a prefect.orion.schemas.core.FlowRun instance as an argument to a Task. This raises a rather misleading error

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

Expectation: The Task object should either a). accept the object as a valid argument or b). raise a more descriptive error TypeError: cannot accept FlowRun instance as Task argument.

Reproduction

from prefect import flow, task
from prefect.context import get_run_context
from prefect.orion.schemas import core


@task()
def some_task(flow_run: core.FlowRun | None) -> None:
    print('This runs if flow_run is None')


@flow()
def my_flow() -> None:
    flow_run_context = get_run_context()

    # this task runs
    some_task(flow_run=None)

    # this task fails
    some_task(flow_run=flow_run_context.flow_run)


my_flow()

Error

root@75811518bc05:/exos# python -m flow
15:41:14.310 | INFO    | prefect.engine - Created flow run 'stoic-bee' for flow 'my-flow'
15:41:14.693 | INFO    | Flow run 'stoic-bee' - Created task run 'this_task_fails-a99f3bf1-0' for task 'this_task_fails'
15:41:14.695 | INFO    | Flow run 'stoic-bee' - Executing 'this_task_fails-a99f3bf1-0' immediately...
15:41:14.943 | INFO    | Task run 'this_task_fails-a99f3bf1-0' - Finished in state Completed()
15:41:15.005 | INFO    | Flow run 'stoic-bee' - Finished in state Completed('All states completed.')
root@75811518bc05:/exos# python -m flow
15:43:33.543 | INFO    | prefect.engine - Created flow run 'truthful-vulture' for flow 'my-flow'
15:43:34.112 | INFO    | Flow run 'truthful-vulture' - Created task run 'some_task-c498e59a-0' for task 'some_task'
15:43:34.120 | INFO    | Flow run 'truthful-vulture' - Executing 'some_task-c498e59a-0' immediately...
This runs if flow_run is None
15:43:34.389 | INFO    | Task run 'some_task-c498e59a-0' - Finished in state Completed()
15:43:34.473 | INFO    | Flow run 'truthful-vulture' - Created task run 'some_task-c498e59a-1' for task 'some_task'
15:43:34.474 | INFO    | Flow run 'truthful-vulture' - Executing 'some_task-c498e59a-1' immediately...
15:43:34.552 | ERROR   | Flow run 'truthful-vulture' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 624, in orchestrate_flow_run
    result = await run_sync(flow_call)
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/asyncutils.py", line 69, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/usr/local/lib/python3.10/dist-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/exos/flow.py", line 19, in my_flow
    some_task(flow_run=flow_run_context.flow_run)
  File "/usr/local/lib/python3.10/dist-packages/prefect/tasks.py", line 367, in __call__
    return enter_task_run_engine(
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 824, in enter_task_run_engine
    return run_async_from_worker_thread(begin_run)
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/asyncutils.py", line 152, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/usr/local/lib/python3.10/dist-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 458, in result
    return self.__get_result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 965, in get_task_call_return_value
    return await future._result()
  File "/usr/local/lib/python3.10/dist-packages/prefect/futures.py", line 237, in _result
    return await final_state.result(raise_on_failure=raise_on_failure, fetch=True)
  File "/usr/local/lib/python3.10/dist-packages/prefect/states.py", line 89, in _get_state_result
    raise MissingResult(
prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.
15:43:34.657 | ERROR   | Flow run 'truthful-vulture' - Finished in state Failed('Flow run encountered an exception. MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.\n')
Traceback (most recent call last):
  File "/usr/lib/python3.10/runpy.py", line 196, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "/usr/lib/python3.10/runpy.py", line 86, in _run_code
    exec(code, run_globals)
  File "/exos/flow.py", line 22, in <module>
    my_flow()
  File "/usr/local/lib/python3.10/dist-packages/prefect/flows.py", line 447, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 162, in enter_flow_run_engine_from_flow_call
    return anyio.run(begin_run)
  File "/usr/local/lib/python3.10/dist-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/usr/local/lib/python3.10/dist-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 242, in create_then_begin_flow_run
    return await state.result(fetch=True)
  File "/usr/local/lib/python3.10/dist-packages/prefect/states.py", line 77, in _get_state_result
    raise await get_state_exception(state)
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 624, in orchestrate_flow_run
    result = await run_sync(flow_call)
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/asyncutils.py", line 69, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/usr/local/lib/python3.10/dist-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/exos/flow.py", line 19, in my_flow
    some_task(flow_run=flow_run_context.flow_run)
  File "/usr/local/lib/python3.10/dist-packages/prefect/tasks.py", line 367, in __call__
    return enter_task_run_engine(
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 824, in enter_task_run_engine
    return run_async_from_worker_thread(begin_run)
  File "/usr/local/lib/python3.10/dist-packages/prefect/utilities/asyncutils.py", line 152, in run_async_from_worker_thread
    return anyio.from_thread.run(call)
  File "/usr/local/lib/python3.10/dist-packages/anyio/from_thread.py", line 49, in run
    return asynclib.run_async_from_thread(func, *args)
  File "/usr/local/lib/python3.10/dist-packages/anyio/_backends/_asyncio.py", line 970, in run_async_from_thread
    return f.result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 458, in result
    return self.__get_result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.10/dist-packages/prefect/engine.py", line 965, in get_task_call_return_value
    return await future._result()
  File "/usr/local/lib/python3.10/dist-packages/prefect/futures.py", line 237, in _result
    return await final_state.result(raise_on_failure=raise_on_failure, fetch=True)
  File "/usr/local/lib/python3.10/dist-packages/prefect/states.py", line 89, in _get_state_result
    raise MissingResult(
prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

Versions

Version:             2.7.11
API version:         0.8.4
Python version:      3.10.9
Git commit:          6b27b476
Built:               Thu, Feb 2, 2023 7:22 PM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         ephemeral
Server:
  Database:          sqlite
  SQLite version:    3.31.1

Additional context

No response