Short answer: those RolledBack states are expected in Prefect 3.x. Even if you don’t explicitly open a transaction, each task runs inside a transaction by default. When a failure happens later in the same transaction scope, Prefect rolls back earlier “staged” tasks and marks them RolledBack. It does not undo your side effects (like copied files) unless you add rollback hooks; the RolledBack state is informational about orchestration, not a guarantee of external undo. Why it happened in your code - You call

_copy_file

and

_rename_file

inside the

copy_file

task. - Then

archive_file()

(a regular function) raises inside the same

copy_file

task. - That exception causes the enclosing transaction (for

copy_file

) to roll back, which propagates a RolledBack state to the child task runs

_copy_file

and

_rename_file

, even though they executed successfully and performed side effects. How to make the earlier tasks end in Success Pick one of these patterns to isolate or eagerly commit the steps you want to keep as Completed: 1) Commit eagerly for the steps you want to “stick” - Wrap the copy/rename section in an eager transaction so they commit immediately and are not rolled back by later failures.

from prefect import task, flow
from prefect.transactions import transaction, CommitMode

@task
def _copy_file(endpoint, ...):
    # shutil.copy(...)

@task
def _rename_file(endpoint, ...):
    # shutil.move(...)

def archive_file(...):
    if not archive_dir_exists():
        raise RuntimeError("Arch dir does not exists")
    # do_archivation()

@task
def copy_file(...):
    # Commit these immediately
    with transaction(commit_mode=CommitMode.EAGER):
        _copy_file(...)
        _rename_file(...)

    # This may fail, but earlier tasks remain Completed
    archive_file()

@flow
def flow1(...):
    copy_file(...)

2) Split into separate transaction scopes at the flow level - Keep copy/rename in their own eager section; run archive after.

@flow
def flow1(...):
    with transaction(commit_mode=CommitMode.EAGER):
        _copy_file(...)
        _rename_file(...)

    # If this fails, the first two remain Completed
    archive_file()

3) Use a subflow to isolate commits - Run copy/rename in a subflow, then archive in the parent flow.

@flow
def copy_and_rename(...):
    _copy_file(...)
    _rename_file(...)

@flow
def flow1(...):
    copy_and_rename(...)   # commits independently
    archive_file(...)      # can fail without rolling back the above

4) Make archive a separate task and do not raise in the parent task - If you want

copy_file

itself to finish in Success, catch the error locally and let a separate task (or the flow) decide whether to fail the run:

from prefect import get_run_logger

@task
def copy_file(...):
    with transaction(commit_mode=CommitMode.EAGER):
        _copy_file(...)
        _rename_file(...)

    try:
        archive_file()
    except Exception as exc:
        get_run_logger().warning(f"Archive failed: {exc}")
        # Do not re-raise here if you want this task to remain Completed

Notes - RolledBack is a Completed-type state used to indicate the transaction rolled back; Prefect does not revert your filesystem changes unless you add explicit

@task.on_rollback

hooks. - If you want real cleanup on failures, add rollback hooks to your tasks to undo side effects. Docs and references - Transactions in Prefect 3.x: Transactions guide - API (CommitMode, transaction): Transactions API - Task API: Task API If you share whether you prefer to keep your current structure (nested tasks) or move orchestration into the flow, I can suggest the minimal change that preserves Completed for the copy/rename steps.

Great goal — you want copy/rename to run exactly once, and if archiving fails, future runs should skip those steps by using Cached results. To guarantee that: 1) Use a cache policy that works across flow runs (avoid RUN_ID in the key) 2) Commit those task results eagerly so a later failure doesn’t roll them back What to change - DEFAULT includes RUN_ID, so caches won’t be reused across flow runs. Use INPUTS (+ optionally TASK_SOURCE) instead, and exclude any non-deterministic parameters (e.g., your

endpoint

) using

- 'endpoint'

. - Wrap the copy/rename section in an eager transaction so the results are committed immediately and survive downstream failures. Example

from prefect import task, flow
from prefect.cache_policies import INPUTS, TASK_SOURCE
from prefect.transactions import transaction, CommitMode
import shutil

# Cache across runs by inputs (excluding 'endpoint'), invalidate if task code changes
COPY_RENAME_CACHE = (INPUTS - 'endpoint') + TASK_SOURCE

@task(persist_result=True, cache_policy=COPY_RENAME_CACHE)
def _copy_file(src: str, dst: str, endpoint: str, **kwargs):
    shutil.copy(src, dst)
    return True  # small persisted result to back the cache

@task(persist_result=True, cache_policy=COPY_RENAME_CACHE)
def _rename_file(src: str, dst: str, endpoint: str, **kwargs):
    shutil.move(src, dst)
    return True

def archive_file(...):
    if not archive_dir_exists(...):
        raise RuntimeError("Arch dir does not exists")
    do_archivation(...)

@flow
def flow1(src: str, tmp_dst: str, final_dst: str, endpoint: str, **kwargs):
    # Commit copy/rename immediately so they are not rolled back
    with transaction(commit_mode=CommitMode.EAGER):
        _copy_file(src, tmp_dst, endpoint, **kwargs)
        _rename_file(tmp_dst, final_dst, endpoint, **kwargs)

    # This may fail; copy/rename will remain Completed and cached
    archive_file(...)

What you’ll see - First run: copy/rename execute and complete; if

archive_file

fails, they still stay Completed because of EAGER. - Next run with the same inputs: copy/rename return Cached and do not execute again. Optional hardening - If you may run the same inputs concurrently (multiple agents), add a lock for SERIALIZABLE isolation to prevent duplicate work:

from prefect.locking.memory import MemoryLockManager
from prefect.transactions import IsolationLevel

COPY_RENAME_CACHE = ((INPUTS - 'endpoint') + TASK_SOURCE).configure(
    isolation_level=IsolationLevel.SERIALIZABLE,
    lock_manager=MemoryLockManager(),
)

Links - Transactions: Transactions guide - Caching concepts: Caching concepts - How to cache workflow steps: How to cache workflow steps - Cache policies API: Cache Policies API If you share the exact parameters that define “the same copy/rename,” I can suggest a precise cache policy (e.g., include file paths, exclude transient flags) to avoid accidental cache misses.

Got it — you’ve turned the copy/rename step into a subflow (

@flow def copy_file_task

). That’s a good move: a subflow commits independently, so if archiving fails later in the parent flow, the copy/rename flow run remains Completed. To also ensure these steps won’t re-run on the next attempt, make sure they’re cached across flow runs. Checklist to meet “copy once, then use Cached”: - Keep copy/rename in a subflow so they’re isolated from the parent failure. - Set persist_result=True on the copy/rename tasks. - Use a cache policy that does not include RUN_ID (so results are reused across runs), e.g.

(INPUTS [+ TASK_SOURCE])

, and exclude only truly non-deterministic/irrelevant params. - Optionally add SERIALIZABLE locking if there’s any chance of concurrent duplicates. Template

from prefect import task, flow
from prefect.cache_policies import INPUTS, TASK_SOURCE
import shutil

# Choose the parameters that define “same work”. Usually: src and dsts.
# Only exclude params that should NOT affect the cache.
COPY_RENAME_CACHE = (INPUTS /* e.g., - 'verbose' */) + TASK_SOURCE

@task(persist_result=True, cache_policy=COPY_RENAME_CACHE)
def _copy_file(src: str, tmp_dst: str):
    shutil.copy(src, tmp_dst)
    return {"copied_to": tmp_dst}

@task(persist_result=True, cache_policy=COPY_RENAME_CACHE)
def _rename_file(tmp_dst: str, final_dst: str):
    shutil.move(tmp_dst, final_dst)
    return {"renamed_to": final_dst}

@flow
def copy_file_task(src: str, tmp_dst: str, final_dst: str):
    _copy_file(src, tmp_dst)
    _rename_file(tmp_dst, final_dst)

def archive_file(...):
    if not archive_dir_exists(...):
        raise RuntimeError("Arch dir does not exists")
    do_archivation(...)

@flow
def main_flow(src: str, tmp_dst: str, final_dst: str):
    # Runs once; subsequent retries/reruns will be Cached with same inputs
    copy_file_task(src, tmp_dst, final_dst)

    # If this fails, the subflow remains Completed and cached
    archive_file(...)

Important notes - If you previously used `cache_policy=DEFAULT - 'endpoint'`: that still includes RUN_ID and won’t cache across runs. Switch to

(INPUTS - '...') + TASK_SOURCE

as shown. - Be careful excluding `endpoint`: if different endpoints mean different locations, keep it in the cache key. - If concurrent runs might hit the same inputs, consider SERIALIZABLE locking to prevent duplicate work. Docs - Caching concepts: Caching concepts - How to cache workflow steps: How to cache workflow steps - Transactions and isolation (for understanding subflow commit behavior): Transactions guide If you share the exact parameters that define “same copy/rename,” I can help tailor the precise

cache_policy

so you get reliable caching without accidental cache misses or false hits.

Short answer: yes—with copy_file turned into a subflow that only does copy/rename, you don’t need to change the transaction commit_mode to prevent rollback. The parent flow’s later failure (in archiving) won’t roll back the subflow’s task states. Key points to be sure this works as you expect: - Apply the cache policy to the actual tasks that do the work (

_copy_file

,

_rename_file

). Flows don’t use cache policies. - Use

persist_result=True

on those tasks. - Use a cross-run cache policy like

INPUTS - 'endpoint'

(optionally add

+ TASK_SOURCE

to invalidate on code changes). - Keep the subflow limited to copy/rename. If anything inside that same subflow could fail after those tasks, then they could still be rolled back; in that case either split the failing step out (recommended) or use an eager transaction inside the subflow. Example structure:

from prefect import task, flow
from prefect.cache_policies import INPUTS, TASK_SOURCE
import shutil

COPY_RENAME_CACHE = (INPUTS - 'endpoint') + TASK_SOURCE  # reuse across runs

@task(persist_result=True, cache_policy=COPY_RENAME_CACHE)
def _copy_file(src, tmp_dst, endpoint):
    shutil.copy(src, tmp_dst)
    return True

@task(persist_result=True, cache_policy=COPY_RENAME_CACHE)
def _rename_file(tmp_dst, final_dst, endpoint):
    shutil.move(tmp_dst, final_dst)
    return True

@flow
def copy_file_task(src, tmp_dst, final_dst, endpoint):
    _copy_file(src, tmp_dst, endpoint)
    _rename_file(tmp_dst, final_dst, endpoint)

def archive_file(...):
    if not archive_dir_exists(...):
        raise RuntimeError("Arch dir does not exists")
    do_archivation(...)

@flow
def main_flow(...):
    copy_file_task(...)     # completes and persists
    archive_file(...)       # may fail; copy/rename remain Completed/Cached

Docs: - Caching concepts: Caching concepts - How to cache workflow steps: How to cache workflow steps - Transactions (why subflows isolate rollback): Transactions guide

Great question — and thanks for the concrete test. Why task1/task2 were not RolledBack in your runner flow - In your example, runner is a subflow that calls two tasks, then raises. - By default, there is no transaction scope in the flow that wraps those tasks. Each task runs in its own transaction and commits when it finishes. The later exception in the flow does not retroactively roll them back. - That’s why you saw task1 and task2 remain Completed, not RolledBack. When will they be RolledBack? - Only if those tasks are inside an active transaction scope that later rolls back. There are two common ways that happens: 1) You explicitly open a transaction in the flow:

from prefect.transactions import transaction

     @flow
     def runner(msg_id: int):
         with transaction():  # default CommitMode.LAZY
             task1(msg_id)   # staged to parent transaction
             task2(msg_id)
             raise RuntimeError("Test exception")  # triggers rollback → tasks marked RolledBack

2) You call tasks from within another task and the parent task raises afterward (your earlier structure). The parent task’s transaction can cause the child task runs to be marked RolledBack when it fails after they’ve completed. So to reconcile with my earlier statement: “they could still be rolled back” applies only if there’s a surrounding transaction scope (explicit

with transaction()

or an enclosing parent task’s transaction) that later fails. A flow by itself does not implicitly create a rollback scope for its tasks. For your “copy once then cache” goal, your current approach (move copy/rename into a subflow and use a cross-run cache policy like

INPUTS - 'endpoint'

on the tasks) is correct and you do not need to change commit_mode. If you were to wrap those tasks in

with transaction()

and keep the default LAZY mode, then a later exception would mark them RolledBack. Docs for reference - Transactions: Transactions guide - Caching concepts: Caching concepts