Short answer: Yes. In Prefect 3.x, a task/asset materialization can be skipped due to a valid cache hit even if the local file it wrote is gone. Prefect does not check your filesystem for existence; it relies on server-side cached state and result persistence. Assumptions: Prefect 3.x. What causes the wrapped function to be skipped Prefect will return a cached “Completed (Cached)” state instead of running the function when all of the following are true: - There is a previously Completed state stored for the task with a matching cache key - That cached entry is not expired (or has no expiration) - Caching is not disabled and no explicit refresh has been requested - The task’s result is persisted (globally or per-task), so the cached value can be returned Key details (source-verified) - Cache lookup and skip logic happens during the transition to RUNNING; if a match is found, Prefect returns the cached Completed state and skips execution: - Cache retrieval/insertion rules: core_policy.py (CacheRetrieval, CacheInsertion) - Only Completed states are cacheable/retrievable; expired caches are ignored - The default cache key (when you do not supply

cache_key_fn

) comes from a cache policy. By default, Prefect uses a compound policy that includes: - Inputs - Task source (code) - Run-scoping component See policy primitives and defaults here: prefect/cache_policies.py - Result persistence must be enabled for cache retrieval to work across process boundaries: - Global:

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

- Or per-task:

persist_result=True

Settings references: - Results settings: prefect/settings/models/results.py - Task settings (caching toggles): prefect/settings/models/tasks.py - Cached runs do not emit asset events. Asset materialization events are suppressed for “Cached” states, so a cached run won’t count as a fresh materialization: - See

AssetContext.emit_asset_events()

early-return for “Cached”: prefect/context.py Important implications for file-writing assets - Prefect does not re-check your filesystem when deciding to reuse a cached result. If the on-disk file has been deleted, Prefect can still skip execution because the server believes a valid cached Completed state exists. - Asset tracking and task caching are separate concerns: - Task caching controls skip/execute behavior. - Asset materialization is event-based and does not retroactively verify file existence. Recommended recovery patterns for stale or inconsistent asset state Pick a pattern that fits your use case; they can be combined. 1) Add a preflight existence/fingerprint check to your cache key - Use a cache key function that returns None (no cache) if the expected file is missing or looks stale. This forces execution when the local state is inconsistent. Example:

from pathlib import Path
from prefect import task
from prefect.cache_policies import CacheKeyFnPolicy, DEFAULT

def file_aware_cache_key(task, inputs, **_):
    path = inputs.get("path")
    if not path or not Path(path).exists():
        return None  # Force a cache miss -> re-execute
    # Otherwise, use the normal default key (inputs + code + run scoping)
    return DEFAULT.compute_key(task, inputs)

@task(persist_result=True, cache_policy=CacheKeyFnPolicy(file_aware_cache_key))
def build_file_asset(path: str) -> str:
    # write the file and return the path
    ...
    return path

Optionally, include a fingerprint (size, mtime, checksum) in the key to invalidate when contents change: ``` def file_fingerprint_key(task, inputs, **_):

path = inputs.get("path") p = Path(path) if not p.exists(): return None fp = (p.stat().st_size, int(p.stat().st_mtime)) # Combine fingerprint with the default policy’s input/code elements base = DEFAULT.compute_key(task, inputs) return f"{base}:{fp}"

2) Set a reasonable cache TTL
- If your files are ephemeral, set `cache_expiration` so Prefect naturally re-runs after a period:

from datetime import timedelta @task(persist_result=True, cache_expiration=timedelta(hours=6)) def build_file_asset(...): ... ``` 3) Force a refresh when needed - Temporarily set

PREFECT_TASKS_REFRESH_CACHE=true

(or the task-level refresh flag) to bypass retrieval and update the cache on the next run. - Changing the task’s source code also invalidates the default cache (TaskSource is part of the key), so bumping a “version” constant in the function is a simple manual bust. 4) Validate at runtime and raise a controlled miss - In the task itself, verify the file exists and re-create it if missing. This trades a bit of runtime overhead for resilience when the environment drifts. 5) Prefer durable storage for assets - If local files are short-lived or workers are ephemeral, write to durable/remote storage (S3/GCS/etc.) and track the remote URI. Local deletion will not invalidate the asset. Docs to learn more - Caching (Prefect 3.x): https://docs.prefect.io/latest/concepts/caching/ - Assets (Prefect 3.x): https://docs.prefect.io/latest/concepts/assets/ Notes on Prefect 2.x - Prefect 2.x did not have the same first-class Assets model; task caching behavior (cache keys, TTL, need for persisted results) was similar in spirit. If you’re still on 2.x and want assets + improved observability, we recommend upgrading to 3.x. If you share how you’re defining the asset (decorators, any cache policy or TTL you’re using), I can tailor a concrete snippet that prevents skips when the file is missing.