Short answer: check the task’s State. A cache hit will put the task run in a state named “Cached” (type “COMPLETED”). You can access the state immediately after you submit the task. Two easy patterns inside a flow: 1) Get a State directly

@task(cache_policy=CachePolicy.DEFAULT)
def compute(x): 
    return x * 2

@flow
def my_flow():
    # Returns a State instead of a Future
    state = compute.submit(1, return_state=True)
    if state.name == "Cached":
        print("cache hit")
    else:
        print("fresh execution")
    result = get_state_result(state)  # extract the value if needed

2) Use a Future, then inspect its state

@flow
def my_flow():
    fut = compute.submit(1)
    fut.wait()  # ensure the task finished
    state = fut.state
    is_cache_hit = state.name == "Cached"
    print("hit" if is_cache_hit else "miss")
    value = fut.result()

Notes: - Calling a task “plain” (e.g.,

val = compute(1)

) returns just the result, not the State. You won’t be able to tell if it was cached from the value alone—prefer

submit(..., return_state=True)

or a Future to inspect the State. - For a cache hit,

state.name == "Cached"

. For a fresh run that sets the cache, you’ll see a normal “Completed” with

state.state_details.cache_key

populated. - The UI will also show the state “Cached”. Prefect logs may include cache messages, but programmatically the State is the reliable source. Docs: - Caching in Prefect 3.x - States - Tasks Imports for the snippets:

from prefect import flow, task
from prefect.cache_policies import CachePolicy
from prefect.states import get_state_result