Short answer: partly. In self-hosted Prefect 3.x there’s built-in, configurable cleanup for events, but not for flow runs. For flow runs you’ll want to run a small “retention” flow on a schedule. What’s automatic today - Events: Prefect automatically deletes old events on a schedule. You can set the retention window with

PREFECT_EVENTS_RETENTION_PERIOD

(default is 7d). - Example:

PREFECT_EVENTS_RETENTION_PERIOD="30d"

- See the Database maintenance guide: Database maintenance (Prefect 3) What you’ll configure for flow runs - There’s no built-in TTL for flow runs, task runs, or their state history. The recommended approach is to run a small maintenance flow that deletes old runs through the API (so related records are cleaned up safely), then deploy it on a schedule. Example: delete terminal-state flow runs older than N days

import asyncio
from datetime import datetime, timedelta, timezone

from prefect import flow, task, get_run_logger
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterStartTime,
    FlowRunFilterState,
    FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType

@task
async def delete_old_flow_runs(days_to_keep: int = 30, batch_size: int = 100):
    logger = get_run_logger()
    cutoff = datetime.now(timezone.utc) - timedelta(days=days_to_keep)

    async with get_client() as client:
        # filter for terminal-state runs started before cutoff
        flow_run_filter = FlowRunFilter(
            start_time=FlowRunFilterStartTime(before_=cutoff),
            state=FlowRunFilterState(
                type=FlowRunFilterStateType(
                    any_=[StateType.COMPLETED, StateType.FAILED, StateType.CANCELLED]
                )
            ),
        )

        total_deleted = 0
        while True:
            runs = await client.read_flow_runs(flow_run_filter=flow_run_filter, limit=batch_size)
            if not runs:
                break

            for fr in runs:
                try:
                    await client.delete_flow_run(fr.id)
                    total_deleted += 1
                except Exception as e:
                    logger.warning(f"Failed to delete flow run {fr.id}: {e}")

            # small pause to be gentle on the API
            await asyncio.sleep(1.0)

        <http://logger.info|logger.info>(f"Retention complete. Total deleted: {total_deleted}")

@flow(name="flow-run-retention")
async def retention_flow(days_to_keep: int = 30, batch_size: int = 100):
    await delete_old_flow_runs(days_to_keep=days_to_keep, batch_size=batch_size)

Then schedule it as a deployment (e.g., daily at 2 AM) and run it with your existing work pool/worker:

prefect deploy retention.py:retention_flow --name daily-retention --schedule "0 2 * * *"

Notes - The filter above uses

start_time

(i.e., runs that actually started). If you also want to remove “never started” runs after some age, say the word and I can tweak the filter. - The API approach safely cascades related records; direct SQL deletes are not recommended unless you know exactly what you’re doing. - One-off cleanups can be done with the CLI for individual runs:

prefect flow-run delete <flow-run-id>

If you share your desired retention window (e.g., “keep 14 days”), which states to keep/delete, and typical daily volume, I can tailor the filter and batching to your setup.