Yes. In Prefect 3.x a cache hit is represented as a Completed state with the name “Cached”. From a flow, you can inspect the task’s state to tell if the result was served from cache. Quick example:

from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import DEFAULT

@task(cache_policy=DEFAULT, cache_expiration=timedelta(hours=1), persist_result=True)
def compute(x: int) -> int:
    return x + 1

@flow
def my_flow():
    fut = compute.submit(1)

    # get the freshest state from the API
    state = fut.fetch_state()  # or fut.get_state() to use the local cached state

    came_from_cache = state.name == "Cached"  # state.type will still be Completed
    print(f"Cache hit? {came_from_cache}")

    return fut.result()

my_flow()

Notes: -

state.name == "Cached"

is the reliable indicator for a cache hit. The state type will still be

Completed

. -

state.state_details.cache_key

tells you the cache key used/configured, but its presence alone does not mean a hit—it’s also set on fresh executions that populate the cache. - If you’ve already resolved the value to a plain Python object (i.e., you didn’t keep the future), you won’t be able to tell from the value alone. Keep the

PrefectTaskRunFuture

or query the API for the task run to inspect its state. Related docs: - States - Cache policies

Great questions — here’s how to choose and how to ensure you’re looking at a final state. When to use get_state vs fetch_state - get_state(): returns the best-known state in the local process without hitting the API. Use this when you’re in the same flow run/process and don’t expect out-of-band changes. - fetch_state(): makes an API round-trip to get the freshest state from the server. Use this if: - The task may have been updated by another process/worker - You rely on the most up-to-date state (e.g., retries, external transitions) - You’re doing cross-process or distributed execution where local state could be stale Do I need wait()? - If you just call result(), you do not need wait() — result() blocks until the task is finished. - Use wait() when you want the final State object (to inspect name/details) without resolving the result yet. Two common patterns 1) You need both the final state (to detect cache) and the result:

fut = compute.submit(1)

state = fut.wait()  # blocks until terminal state; returns a State
is_cached = state.name == "Cached"
value = fut.result()  # returns immediately now; already finished

2) You only called result() and then want to check if it was cached:

fut = compute.submit(1)
value = fut.result()  # blocks until done
is_cached = fut.get_state().name == "Cached"  # local state is fine in-process
# Or for freshness across processes:
# is_cached = fut.fetch_state().name == "Cached"

Extras -

state.state_details.cache_key

shows the cache key involved, but its presence alone doesn’t guarantee a hit; use

state.name == "Cached"

to detect a cache hit. The state type remains Completed. Docs for reference: - Futures API - States - Cache policies