It sounds like two different things are being written: - Cache records: controlled by

CachePolicy(key_storage=...)

- Persisted results: controlled separately by your task/flow

result_storage

(and whether

persist_result

is enabled) So if you changed

key_storage

, new cache records will go to that new folder — but: - Old cache records will remain in the old folder (Prefect doesn’t move them automatically) - If you’re persisting results, those files will still go to the default result storage unless you change that too How to change the cache record location - Set

key_storage

on the cache policy you pass to your tasks. You can use a path or a filesystem:

from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem

# Option A: plain path
my_cache = CachePolicy(key_storage="/mnt/prefect/cache")

# Option B: LocalFileSystem
# my_cache = CachePolicy(key_storage=LocalFileSystem(path="/mnt/prefect/cache"))

@task(cache_policy=my_cache)
def compute(...):
    ...

If you want all cached tasks to write to the same place, make sure every task that uses caching gets this same

CachePolicy

. Otherwise some tasks will still write to the default location. If you also want to move persisted results - Set result storage at the flow level so it applies to tasks, and enable persistence if needed:

from prefect import flow

@flow(result_storage="/mnt/prefect/results")  # set where results are stored
def my_flow():
    ...

Housekeeping - It’s safe to delete old cache records if you don’t need them anymore (it will just force recomputation the next time those keys are requested). - If you’re seeing files in two locations, it’s usually: - legacy cache records in the old folder, plus new cache records in your new

key_storage

- and/or cache records in one location and persisted results in another Docs for reference - Caching in Prefect 3 - Results and result storage Can you share the two paths you’re seeing and a snippet showing how you configured

CachePolicy

? That’ll let me confirm whether it’s cache vs result files or a mix of default and custom policies.

Short answer: In Prefect 3.x,

result_storage

can be a filesystem block, a saved block reference (UUID), or a pathlib.Path. If passing a path is erroring, the most reliable fix is to pass a filesystem object explicitly. Try one of these patterns: - Local directory via a filesystem object

from prefect import flow
from prefect.filesystems import LocalFileSystem

@flow(result_storage=LocalFileSystem(basepath="/tmp/prefect-results"), persist_result=True)
def my_flow():
    return 42

- Local directory via Path

from pathlib import Path
from prefect import flow

@flow(result_storage=Path("/tmp/prefect-results"), persist_result=True)
def my_flow():
    return 42

- Saved block you can reuse across runs/environments

from prefect.filesystems import LocalFileSystem

# one-time setup
LocalFileSystem(basepath="/tmp/prefect-results").save("results-local", overwrite=True)

# later
from prefect import flow
@flow(result_storage=LocalFileSystem.load("results-local"), persist_result=True)
def my_flow():
    return 42

Notes: - Set

persist_result=True

on the flow (and/or tasks) if you actually want values written to storage. - Ensure the directory exists and your worker has write access if you use a local path. - For remote storage, pass the appropriate filesystem block (e.g., S3/GCS/Azure from the integrations) or a saved block reference. If this still raises an error, can you share: - Your Prefect version (

prefect version

) - The exact code snippet where you set

result_storage

- The full error traceback That’ll help me pinpoint whether it’s a parsing issue (e.g., Pydantic union) or an environment/permissions problem.