Great question — and thanks for specifying Prefect 2.20.16. There are a few reliable patterns to clean up files on a mounted volume in Prefect 2.x: Option A: A dedicated cleanup task called in a finally block (runs even if the flow’s tasks fail)

from pathlib import Path
import shutil
import time
from prefect import flow, task, get_run_logger

@task
def do_work(volume_path: str):
    logger = get_run_logger()
    # ... your logic that writes to the volume ...
    <http://logger.info|logger.info>(f"Writing files to {volume_path}")

@task
def cleanup_volume(volume_path: str, remove_older_than_seconds: int | None = None):
    """
    Deletes files under volume_path. If remove_older_than_seconds is set,
    only deletes files older than that threshold.
    """
    logger = get_run_logger()
    root = Path(volume_path)
    if not root.exists():
        <http://logger.info|logger.info>(f"Volume path {volume_path} does not exist, nothing to clean")
        return

    now = time.time()
    deleted = 0

    for p in root.rglob("*"):
        try:
            # Skip directories; we’ll remove empty dirs later
            if p.is_file():
                if remove_older_than_seconds is None:
                    p.unlink(missing_ok=True)
                    deleted += 1
                else:
                    age = now - p.stat().st_mtime
                    if age >= remove_older_than_seconds:
                        p.unlink(missing_ok=True)
                        deleted += 1
        except Exception as exc:
            logger.warning(f"Failed to delete {p}: {exc}")

    # Optionally prune empty directories
    for d in sorted([p for p in root.rglob("*") if p.is_dir()], reverse=True):
        try:
            if not any(d.iterdir()):
                d.rmdir()
        except Exception as exc:
            logger.debug(f"Failed to remove dir {d}: {exc}")

    <http://logger.info|logger.info>(f"Cleanup complete under {volume_path}; deleted {deleted} files")

@flow
def my_flow(volume_path: str = "/data"):
    try:
        do_work(volume_path)
        # ... other tasks ...
    finally:
        # Ensures cleanup runs even if upstream tasks fail
        cleanup_volume(volume_path)

Option B: Use flow state-change hooks so cleanup runs on success, failure, and/or crash - Hooks in Prefect 2.x let you run code when a flow finishes, fails, or crashes. - You can trigger your cleanup task from these hooks so it runs even when the main flow code can’t continue. - Docs: State-change hooks (Prefect 2) Example using on_completion and on_crashed:

from prefect import flow, task
from prefect import get_run_logger

@task
def cleanup_volume(volume_path: str):
    # same implementation as above (or simplified)
    ...

def run_cleanup_on_completion(flow, flow_run, state, volume_path: str = "/data"):
    # Called when the flow enters a Completed or Failed state (after retries)
    cleanup_volume.submit(volume_path)

def run_cleanup_on_crashed(flow, flow_run, state, volume_path: str = "/data"):
    # Called when the flow is marked Crashed (e.g., process killed)
    cleanup_volume.submit(volume_path)

@flow(
    on_completion=[run_cleanup_on_completion],
    on_crashed=[run_cleanup_on_crashed],
)
def my_flow(volume_path: str = "/data"):
    # normal flow logic
    ...

Notes on hooks: - on_completion runs after the flow finishes (success or failure). - on_failure runs only after final failure (after retries are exhausted). - on_crashed runs when Prefect marks the run as crashed (useful if the worker process died). - Hooks are plain functions; to run Prefect tasks from a hook, call

cleanup_volume.submit(...)

. When to use which - If you want cleanup to run no matter what during normal execution: Option A (try/finally) is simple and reliable. - If you want cleanup to also run when the process crashes or is killed: add Option B hooks, especially

on_crashed

, to cover abrupt termination. Infrastructure specifics

- Docker: bind mounts (e.g.,

./data:/app/data

) persist on the host; you must delete unneeded files explicitly. The container’s own filesystem is ephemeral and goes away when the container stops. - Kubernetes: - emptyDir volumes are cleaned up when the pod terminates. - PVC/hostPath volumes persist and require explicit cleanup. - If you rely on Kubernetes GC, you can also set TTL on finished jobs (infra-level, outside Prefect). A couple of clarifying questions to tailor the solution: - Where are you running your flows (local, Docker, Kubernetes)? If Kubernetes, are you using PVCs or emptyDir? - Do you want cleanup to run on every completion, only on failure, or also on crashes? - What path(s) are you cleaning and what qualifies as “unneeded” (e.g., delete everything vs. older-than-X)? If you’d like, share a bit more about your runtime environment and volume mount path and I’ll adapt the code to your setup.

Short answer: look at the Kubernetes Job/Pod spec that Prefect creates, or the work pool’s base job template. Here are a few easy ways: Option 1: Inspect a running pod in your cluster - Find the pod for a flow run (replace labels/namespace as needed):

kubectl get pods -n <namespace> -l <http://prefect.io/flow-run-id|prefect.io/flow-run-id>

- Describe or dump the pod YAML and look at `.spec.volumes`:

kubectl describe pod <pod-name> -n <namespace>
# or
kubectl get pod <pod-name> -n <namespace> -o yaml

What to look for in .spec.volumes: - If you see entries like:

volumes:
- name: scratch
  emptyDir: {}

…that’s an emptyDir. - If you see:

volumes:
- name: data
  persistentVolumeClaim:
    claimName: my-pvc

…that’s a PVC. Option 2: Inspect your Prefect work pool’s base job template - In the UI: Work Pools -> select your Kubernetes pool -> Base job template. Look for

spec.template.spec.volumes

and

containers[0].volumeMounts

. - Via CLI:

prefect work-pool inspect "<your-pool-name>"

Then scan the printed base job template for

volumes:

and see if they’re

emptyDir: {}

or

persistentVolumeClaim:

. Option 3: Inspect the deployment to see pool and job variables - In the UI: Deployments -> select your deployment -> Overview to see which work pool it targets; check “Job variables” for any overrides that might add volumes. - Via CLI:

prefect deployment inspect <flow_name>/<deployment_name>

This shows the work pool name and any job variables. If the deployment overrides volumes, you’ll see them here. Tip: if you don’t see any `volumes`/`volumeMounts` at all, your flow is likely just using the container’s filesystem (ephemeral) and not a volume. For persistent data across pods, you’d see a PVC; for scratch space within a pod lifecycle, you’d see an emptyDir. If you can share your work pool name/namespace, I can suggest the exact kubectl commands with the labels Prefect applies, or help interpret a snippet of your job/pod YAML.

Great — then you can tell which kind of volume you have by looking at the entries under

.spec.volumes

and matching them to

.spec.containers[*].volumeMounts

by

name

. What to look for in `.spec.volumes`: - emptyDir (ephemeral, auto-cleaned when the pod terminates)

volumes:
- name: scratch
  emptyDir: {}               # or emptyDir: { sizeLimit: "2Gi" }

- PVC (persists after the pod; you must clean/remove files yourself)

volumes:
- name: data
  persistentVolumeClaim:
    claimName: my-pvc

- Other common types you might see: -

hostPath:

(persists on the node) -

configMap:

/

secret:

(read-only, not for scratch data) -

projected:

(combines multiple sources) Then map each

volumes[*].name

to the corresponding

volumeMounts[*].name

to see which mountPath is which:

containers:
- name: flow
  volumeMounts:
  - name: data
    mountPath: /data        # This path corresponds to the volume named "data"

Cleanup implications: - If it’s emptyDir: Kubernetes deletes it automatically when the pod ends — you usually don’t need a cleanup task. - If it’s PVC or hostPath: contents persist and you should clean them explicitly (e.g., Prefect cleanup task). Example cleanup task you can call from your flow (or from state-change hooks like

on_completion

/

on_crashed

):

from pathlib import Path
from prefect import task, get_run_logger

@task
def cleanup_mount(mount_path: str, delete_everything: bool = True):
    logger = get_run_logger()
    root = Path(mount_path)
    if not root.exists():
        <http://logger.info|logger.info>(f"{mount_path} does not exist; nothing to clean.")
        return

    # simple "delete everything" strategy; add filters/age checks as needed
    for p in root.rglob("*"):
        try:
            if p.is_file():
                p.unlink(missing_ok=True)
        except Exception as e:
            logger.warning(f"Failed to delete file {p}: {e}")

    # prune empty dirs
    for d in sorted([p for p in root.rglob("*") if p.is_dir()], reverse=True):
        try:
            if not any(d.iterdir()):
                d.rmdir()
        except Exception:
            pass

    <http://logger.info|logger.info>(f"Cleanup complete under {mount_path}")

If you’d like, paste just the

volumes:

block (and the related

volumeMounts:

for the container), and I’ll confirm whether it’s emptyDir or PVC and suggest the best cleanup approach for your setup.

Got it — since it’s a persistentVolumeClaim (PVC), anything your flow writes will persist after the pod ends, so you’ll want to delete it explicitly. Recommended pattern - Write all run artifacts under a run-scoped subdirectory of the PVC mount (e.g., /mnt/data/<flow_run_id>). - Always call a cleanup task in a try/finally so it runs even when upstream tasks fail. - Optionally add flow hooks for “normal” completion; for true crash scenarios, consider a separate cleanup deployment (details below). Example: run-scoped dir + cleanup task

from pathlib import Path
import shutil
from prefect import flow, task, get_run_logger
from prefect.runtime import flow_run as rt_flow_run  # gives you the current flow run id

@task
def do_work(run_dir: str):
    logger = get_run_logger()
    p = Path(run_dir)
    p.mkdir(parents=True, exist_ok=True)
    # Write your files under run_dir
    (p / "example.txt").write_text("some data")
    <http://logger.info|logger.info>(f"Wrote files under {run_dir}")

@task
def cleanup_run_dir(base_mount: str, run_id: str):
    """
    Removes the per-run directory on the PVC.
    """
    logger = get_run_logger()
    run_dir = Path(base_mount) / run_id
    if run_dir.exists():
        try:
            shutil.rmtree(run_dir, ignore_errors=True)
            <http://logger.info|logger.info>(f"Removed run directory: {run_dir}")
        except Exception as e:
            logger.warning(f"Failed to remove {run_dir}: {e}")
    else:
        <http://logger.info|logger.info>(f"Run directory not found: {run_dir}")

@flow
def my_flow(base_mount: str = "/mnt/data"):
    run_id = rt_flow_run.id
    run_dir = str(Path(base_mount) / run_id)

    try:
        do_work(run_dir)
        # ... other tasks that read/write under run_dir ...
    finally:
        # Ensure cleanup runs even if tasks fail
        cleanup_run_dir.submit(base_mount, run_id)

Why a run-scoped directory? - It keeps each run isolated and makes cleanup a simple, safe

rm -rf

of that single folder. - Use

prefect.runtime.flow_run.id

so the directory name is unique per flow run. Optional: flow hooks - You can also attach hooks so cleanup runs on completion/failure. Example:

def run_cleanup_on_completion(flow, flow_run, state, base_mount="/mnt/data"):
    cleanup_run_dir.submit(base_mount, flow_run.id)

@flow(on_completion=[run_cleanup_on_completion])
def my_flow(...):
    ...

Note: Hooks run in the flow process when the state changes; they won’t trigger if the process hard-crashes and can’t execute the hook. More robust cleanup for crash scenarios - If you need cleanup even when the original pod is killed, run a separate “cleanup” deployment that mounts the same PVC and deletes the run directory. Trigger it via: - Prefect Cloud Automation on flow-run Completed/Failed/Crashed, or - A Kubernetes CronJob that periodically prunes old run directories (e.g., older than N hours). - Ensure the cleanup deployment’s job template mounts the same

persistentVolumeClaim

with the same

claimName

and

mountPath

. Kubernetes tips - Verify your container user has permission to delete files on the PVC (securityContext/runAsUser). - Prefer deleting a known subdirectory (run-scoped) instead of the whole mount path to avoid accidental data loss. - If you use

subPath

in your volumeMount, point the cleanup at that exact subPath. If you share: - the PVC mountPath in your container (e.g., /mnt/data), and - whether you’re on Prefect Cloud (for Automations), I can tailor the cleanup task and, if helpful, a separate “cleanup deployment” template that mounts your PVC and prunes run directories.

Perfect — since you’re on a PVC and your flow writes under

/dbt

, the safest pattern is to isolate each run in its own subdirectory on the PVC and delete that folder at the end. Below is a drop-in version of your flow that: - Uses a run-scoped path

/dbt/<flow_run_id>

so cleanup is just removing that directory - Ensures cleanup runs with a try/finally - Keeps your existing tasks intact

import os
import shutil
from pathlib import Path

from prefect import flow, task
from prefect.blocks.system import Secret
from prefect.runtime import flow_run as rt_flow_run  # gives current flow run id

try:
    import pydantic
    from pydantic.v1 import SecretStr as V1SecretStr
    pydantic.SecretStr = V1SecretStr
except Exception:
    pass

from prefect_dbt.cli.credentials import DbtCliProfile

from flows.dbt.tasks import dbt_cmd, pull_dbt_repo, send_freshness_alert
from liva.prefect import trigger_deployment


@task
def cleanup_path(path: str):
    p = Path(path)
    if p.exists():
        # Be careful: only remove the run-scoped directory, not the entire mount
        shutil.rmtree(p, ignore_errors=True)

@flow
def dbt_anon():
    git_token = Secret.load("github-access-token").get()

    # Use a run-scoped directory on the PVC: /dbt/<flow_run_id>
    base_mount = "/dbt"  # this is your PVC mountPath in the pod
    run_id = rt_flow_run.id
    dbt_dir = os.path.join(base_mount, run_id)

    env = os.getenv("ENVIRONMENT")

    # Ensure run directory exists before cloning / pulling
    Path(dbt_dir).mkdir(parents=True, exist_ok=True)

    try:
        # Clone/pull repo into the run-scoped path
        pull_dbt_repo.submit(git_token, dbt_dir).wait()

        workdir = os.path.join(dbt_dir, "liva_postgres")
        dbt_cli_profile = DbtCliProfile.load("analytics-anon")

        dbt_cmd("dbt deps", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt seed", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt snapshot", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        anon_run = dbt_cmd("dbt run --exclude tag:stats", dbt_cli_profile=dbt_cli_profile, workdir=workdir)

        # trigger stats tables only
        trigger_deployment.submit(
            flow_name="dbt-stats",
            deployment_name="trigger-dbt-stats",
            wait_for=[anon_run],
        )

        dbt_cmd("dbt deps", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt test --exclude tag:stats", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        if env == "production":
            freshness_output = dbt_cmd("dbt source freshness", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
            send_freshness_alert(freshness_output, "analytics-bot-webhook")

    finally:
        # Always clean up the run-scoped directory on the PVC
        cleanup_path.submit(dbt_dir)

Notes and options: - This expects

pull_dbt_repo(git_token, dbt_dir)

to clone into the given

dbt_dir

. If your task currently assumes a fixed path, update it to honor the directory argument (or I can help refactor it). - Using a run-scoped directory avoids collisions between concurrent runs and makes cleanup safe. - If you want an extra belt-and-suspenders on normal completion/failure, you can also add flow hooks: - State-change hooks (Prefect 2.x) - For crash scenarios (pod killed), the

finally

block won’t run. If you need guaranteed cleanup in those cases, consider: - A small “cleanup” deployment that mounts the same PVC and deletes stale run directories, triggered by a Prefect Cloud Automation on flow run state changes, or - A lightweight Kubernetes CronJob that prunes directories older than N hours. Reference: - Runtime context (for

prefect.runtime.flow_run.id

)

If you share the current

pull_dbt_repo

implementation (or how it decides where to clone), I can tailor the minimal change needed so it clones into the run-scoped

dbt_dir

.

Great — your

pull_dbt_repo

already takes a directory and wipes it before cloning, so it fits perfectly with a run-scoped directory on the PVC. You don’t need to change that task; just pass it a unique per-run path and then clean that path up in a finally. Here’s a minimal, safe refactor of your flow:

import os
import shutil
from pathlib import Path

from prefect import flow, task
from prefect.blocks.system import Secret
from prefect.runtime import flow_run as rt_flow_run

try:
    import pydantic
    from pydantic.v1 import SecretStr as V1SecretStr
    pydantic.SecretStr = V1SecretStr
except Exception:
    pass

from prefect_dbt.cli.credentials import DbtCliProfile

from flows.dbt.tasks import dbt_cmd, pull_dbt_repo, send_freshness_alert
from liva.prefect import trigger_deployment


@task(name="Cleanup run dir")
def cleanup_path(path: str, base_mount: str = "/dbt"):
    # Safety guard: only delete a subdir inside the mount and only if path includes a run id
    p = Path(path).resolve()
    base = Path(base_mount).resolve()
    if not str(p).startswith(str(base)) or p == base:
        # Do nothing if path is outside the mount or equals the mount root
        return
    shutil.rmtree(p, ignore_errors=True)

@flow
def dbt_anon():
    git_token = Secret.load("github-access-token").get()
    env = os.getenv("ENVIRONMENT")

    base_mount = "/dbt"  # PVC mountPath in your pod
    run_id = rt_flow_run.id
    dbt_dir = os.path.join(base_mount, run_id)  # /dbt/<flow_run_id>

    Path(dbt_dir).mkdir(parents=True, exist_ok=True)

    try:
        # Clone into the run-scoped directory
        pull_dbt_repo.submit(git_token, dbt_dir).wait()

        workdir = os.path.join(dbt_dir, "liva_postgres")
        dbt_cli_profile = DbtCliProfile.load("analytics-anon")

        dbt_cmd("dbt deps", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt seed", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt snapshot", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        anon_run = dbt_cmd("dbt run --exclude tag:stats", dbt_cli_profile=dbt_cli_profile, workdir=workdir)

        # trigger stats tables only
        trigger_deployment.submit(
            flow_name="dbt-stats",
            deployment_name="trigger-dbt-stats",
            wait_for=[anon_run],
        )

        dbt_cmd("dbt deps", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt test --exclude tag:stats", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        if env == "production":
            freshness_output = dbt_cmd("dbt source freshness", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
            send_freshness_alert(freshness_output, "analytics-bot-webhook")

    finally:
        # Always clean up the run-scoped directory on the PVC
        cleanup_path.submit(dbt_dir, base_mount=base_mount)

Why this works well on a PVC - Each run gets its own folder (

/dbt/<flow_run_id>

), so concurrent runs don’t collide. - Cleanup is a simple, safe delete of that run’s folder in a finally block. - The cleanup task includes a guard so you can’t accidentally delete the entire mount. Your existing

pull_dbt_repo

fits this pattern as-is:

@task(name="Clone DBT")
def pull_dbt_repo(git_token, directory):
    shutil.rmtree(directory, ignore_errors=True)  # Delete folder on run
    callbacks = pygit2.RemoteCallbacks(pygit2.UserPass(git_token, "x-oauth-basic"))
    pygit2.clone_repository("<https://github.com/x/y.git>", directory, callbacks=callbacks)

If you need cleanup even when the pod is killed - The finally block won’t run if the pod is terminated abruptly. If you want belt-and-suspenders: - Create a tiny “cleanup” flow that mounts the same PVC and deletes

/dbt/<flow_run_id>

. - Trigger it via a Prefect Automation on flow-run state changes (Completed/Failed/Crashed), or run a small Kubernetes CronJob that prunes old run directories periodically. - Docs for context:

- State-change hooks (run on normal completion/failure): https://docs-2.prefect.io/latest/guides/state-change-hooks/ - Runtime context for `prefect.runtime.flow_run.id`: https://docs-2.prefect.io/latest/guides/runtime-context/ If you share the volumeMount (mountPath, subPath if any) I can double-check the safety guard and path handling for your exact setup.

Short answer: it will clean up the space your flow created, but it can’t guarantee you’ll never hit “No space left on device” in every scenario. What your refactor does well - By cloning into /dbt/<flow_run_id> and deleting that directory in a finally block, you reclaim space from completed runs. This prevents the PVC from growing indefinitely due to successful/failed-but-executed runs. - Because your dbt workdir is inside that run directory, dbt’s target/logs live under the same folder and get removed too. Where “No space left” can still happen - Crash/kill before cleanup: if the pod is terminated abruptly, the finally block won’t run and the run directory will remain. Mitigation: add a periodic “sweeper” (cleanup flow or Kubernetes CronJob) that deletes stale /dbt/<run_id> folders older than N hours. - Concurrent runs: multiple large runs can fill the PVC before any of them reach cleanup. Mitigation: limit parallelism, reduce per-run footprint, or add intermediate pruning. - Writes outside the run directory: make sure everything writes under /dbt/<flow_run_id> (including dbt target/logs and any temp files). If anything writes to another path on the PVC (or to hostPath), cleanup won’t touch it. - Permissions: if the container user can’t delete files on the PVC, cleanup may silently fail. Ensure securityContext (runAsUser/runAsGroup/fsGroup) allows deletion. Quick sanity checks to add - Log disk usage before and after cleanup to verify space is reclaimed:

from prefect import get_run_logger
import shutil

logger = get_run_logger()
total, used, free = shutil.disk_usage("/dbt")
<http://logger.info|logger.info>(f"PVC before: free={free/1e9:.2f}GB used={used/1e9:.2f}GB total={total/1e9:.2f}GB")
# ... run work ...
# in finally, after cleanup
total, used, free = shutil.disk_usage("/dbt")
<http://logger.info|logger.info>(f"PVC after: free={free/1e9:.2f}GB used={used/1e9:.2f}GB total={total/1e9:.2f}GB")

Optional: add a sweeper for crash cases - A tiny flow/deployment that mounts the same PVC and removes /dbt/* directories older than N hours. Trigger it via a Prefect Cloud Automation on flow run state changes or run on a schedule (CronJob). This covers the “pod died before finally” scenario. If you confirm: - your PVC mountPath is /dbt (no subPath), and - dbt only writes under that run-scoped workdir, then yes: your new flow will clean up after itself and substantially reduce the risk of “No space left on device.” For full protection, pair it with a periodic sweeper and keep an eye on concurrency.

Absolutely — here’s a complete version of your flow with disk usage logging before the work starts and after cleanup. The “after” log waits for the cleanup task to finish so you can verify space was reclaimed.

import os
import shutil
from pathlib import Path

from prefect import flow, task, get_run_logger
from prefect.blocks.system import Secret
from prefect.runtime import flow_run as rt_flow_run

try:
    import pydantic
    from pydantic.v1 import SecretStr as V1SecretStr
    pydantic.SecretStr = V1SecretStr
except Exception:
    pass

from prefect_dbt.cli.credentials import DbtCliProfile

from flows.dbt.tasks import dbt_cmd, pull_dbt_repo, send_freshness_alert
from liva.prefect import trigger_deployment


@task(name="Cleanup run dir")
def cleanup_path(path: str, base_mount: str = "/dbt"):
    # Safety guard: only delete a subdir inside the mount and never the mount root
    p = Path(path).resolve()
    base = Path(base_mount).resolve()
    if not str(p).startswith(str(base)) or p == base:
        return
    shutil.rmtree(p, ignore_errors=True)


@flow
def dbt_anon():
    logger = get_run_logger()
    git_token = Secret.load("github-access-token").get()
    env = os.getenv("ENVIRONMENT")

    base_mount = "/dbt"  # PVC mountPath in your pod
    run_id = rt_flow_run.id
    dbt_dir = os.path.join(base_mount, run_id)  # /dbt/<flow_run_id>

    Path(dbt_dir).mkdir(parents=True, exist_ok=True)

    # Quick sanity check: log disk usage before work
    total, used, free = shutil.disk_usage(base_mount)
    <http://logger.info|logger.info>(
        f"PVC {base_mount} BEFORE: free={free/1e9:.2f}GB used={used/1e9:.2f}GB total={total/1e9:.2f}GB"
    )

    cleanup_future = None
    try:
        # Clone into the run-scoped directory
        pull_dbt_repo.submit(git_token, dbt_dir).wait()

        workdir = os.path.join(dbt_dir, "liva_postgres")
        dbt_cli_profile = DbtCliProfile.load("analytics-anon")

        dbt_cmd("dbt deps", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt seed", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt snapshot", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        anon_run = dbt_cmd("dbt run --exclude tag:stats", dbt_cli_profile=dbt_cli_profile, workdir=workdir)

        # trigger stats tables only
        trigger_deployment.submit(
            flow_name="dbt-stats",
            deployment_name="trigger-dbt-stats",
            wait_for=[anon_run],
        )

        dbt_cmd("dbt deps", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        dbt_cmd("dbt test --exclude tag:stats", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
        if env == "production":
            freshness_output = dbt_cmd("dbt source freshness", dbt_cli_profile=dbt_cli_profile, workdir=workdir)
            send_freshness_alert(freshness_output, "analytics-bot-webhook")

    finally:
        # Always clean up the run-scoped directory on the PVC
        cleanup_future = cleanup_path.submit(dbt_dir, base_mount=base_mount)
        # Wait for cleanup to complete before measuring disk usage
        cleanup_future.wait()

        # Quick sanity check: log disk usage after cleanup
        total, used, free = shutil.disk_usage(base_mount)
        <http://logger.info|logger.info>(
            f"PVC {base_mount} AFTER:  free={free/1e9:.2f}GB used={used/1e9:.2f}GB total={total/1e9:.2f}GB"
        )

This will emit two log lines per run: - BEFORE: free/used/total at the start - AFTER: free/used/total after the cleanup task has finished If you want an additional check, you can also log right after the clone or after dbt steps to see how much each phase consumes.