File ~/model/.venv/lib/python3.10/site-packages/prefect/states.py:509, in get_state_exception(state)
    507     default_message = "Run cancelled."
    508 else:
--> 509     raise ValueError(f"Expected failed or crashed state got {state!r}.")
    511 if isinstance(state.data, ResultRecord):
    512     result = state.data.result

ValueError: Expected failed or crashed state got Running(message='', type=RUNNING, result=None).

from time import sleep
from prefect import flow, task
from prefect_dask import DaskTaskRunner

@flow(task_runner=DaskTaskRunner(address="localhost:8786"))
def wait_flow():
    @task
    def wait():
        sleep(30)
        return True
    result = wait.submit()
    return result

if __name__ == "__main__":
    wait_flow()

distributed.scheduler.KilledWorker: Attempted to run task slow_task-a25511e480b7951003007c0155e3f56c on 1 different workers, but all those workers died while running it. The last worker that attempt to run the task was tcp://127.0.0.1:60949. Inspecting worker logs is often a good next step to diagnose what went wrong. For more information see https://distributed.dask.org/en/stable/killed.html.

Version:             3.2.7
API version:         0.8.4
Python version:      3.10.16
Git commit:          d4d9001e
Built:               Fri, Feb 21, 2025 7:39 PM
OS/Arch:             darwin/arm64
Profile:             default
Server type:         cloud
Pydantic version:    2.9.2
Integrations:
  prefect-dask:      0.3.2.dev1046+gbe1ba636e4.d20250305
  prefect-gcp:       0.6.2
  prefect-kubernetes: 0.5.3