Great question — and you explained it well. There are a few solid patterns in Prefect 3.x to run a “final decision” after all retries are exhausted. The right one depends on whether you want: - the original task run to be marked Completed if your final check passes, or - you’re okay with the task showing Failed but the flow still succeeding. Here are the two most useful approaches. Option 1 — Do the final check inside the task on the last attempt - Let Prefect handle all retries by re-raising on non-final attempts - On the last attempt only, run your custom check and either return a value (marking the task Completed) or re-raise (keeping it Failed)

from prefect import task, flow, runtime

MAX_RETRIES = 144
TOTAL_ATTEMPTS = MAX_RETRIES + 1  # first run + retries

def final_ok_check(exc) -> bool:
    # Your custom logic — return True to accept as success
    return "benign" in str(exc).lower()

def do_the_thing():
    # Your core logic
    ...

@task(retries=MAX_RETRIES, retry_delay_seconds=5 * 60)
def my_task():
    attempt = runtime.task_run.run_count  # starts at 1
    try:
        return do_the_thing()
    except Exception as exc:
        if attempt < TOTAL_ATTEMPTS:
            # Not the last attempt — let Prefect retry
            raise
        # Last attempt — decide if we accept this as “success”
        if final_ok_check(exc):
            return "accepted"  # task will be Completed
        # Otherwise, keep task Failed
        raise

@flow
def my_flow():
    return my_task()

Notes: - This is the only way to end with a Completed task even if all prior attempts failed. - Prefect cannot change a Failed state to Completed from an on_failure hook — hooks are side effects only. Option 2 — Keep the task’s final state, but decide at the flow level whether to proceed/succeed If you’re okay with the task showing Failed but want to let the flow proceed/succeed based on your custom check, handle it in the flow: Pattern A: Use return_state=True to inspect the final state

from prefect import flow, task

@task(retries=144, retry_delay_seconds=5 * 60)
def my_task():
    ...

def final_ok_check(exc) -> bool:
    ...

@flow
def parent():
    state = my_task(return_state=True)  # don’t raise on failure
    if state.is_failed():
        exc = state.result(raise_on_failure=False)
        if final_ok_check(exc):
            return "accepted"  # flow succeeds
        # Propagate the original failure to fail the flow
        state.result()  # raises
    return state.result()

Pattern B: Use allow_failure for a clean downstream “decider” task

from prefect import flow, task, allow_failure

def final_ok_check(exc) -> bool:
    ...

@task(retries=144, retry_delay_seconds=5 * 60)
def my_task():
    ...

@task
def decide(val_or_exc):
    if isinstance(val_or_exc, Exception):
        if final_ok_check(val_or_exc):
            return "accepted"  # flow can succeed
        # Re-raise to fail the flow (or return a sentinel and keep going)
        raise val_or_exc
    return val_or_exc

@flow
def parent():
    res = my_task()
    return decide(allow_failure(res))

A couple of clarifications and tips: - retry_condition_fn is great for deciding whether to retry on each failure, but it cannot convert a final failure into a success. - on_failure hooks run after retries are exhausted and the task enters Failed, but they cannot change the task’s state — use them only for side effects (logging, alerts, etc.). - You can branch logic by attempt number inside a task using runtime.task_run.run_count (starts at 1). Quick question to tailor this: Do you want the original task to appear as Completed if your final check passes, or is it acceptable for the task to remain Failed while the flow still succeeds based on that check? If you share a bit of pseudo-code for the final check, I can wire it into whichever pattern you prefer.

Short answer: you’re good — each mapped child task run has its own independent retry counter. Your “last-attempt” logic will execute per item, not shared across the whole map. Details: - task.map([...]) creates a separate task run for each item. - Each mapped run tracks its own run_count (attempt number) and applies your retries/retry_delay_seconds independently. - Inside the task, runtime.task_run.run_count (or get_run_context().task_run.run_count) refers to the current mapped child’s attempt number. It starts at 1 and increments on retries for that specific item only. - Your “final check on last attempt” pattern will therefore run separately for each mapped item. Example pattern with mapping:

from prefect import task, flow, runtime

MAX_RETRIES = 144
TOTAL_ATTEMPTS = MAX_RETRIES + 1

def final_ok_check(exc, x) -> bool:
    # Custom logic per item x
    return "benign" in str(exc).lower()

@task(retries=MAX_RETRIES, retry_delay_seconds=5 * 60)
def my_task(x):
    attempt = runtime.task_run.run_count
    try:
        # your main logic using x
        raise RuntimeError(f"Transient for {x}")
    except Exception as exc:
        if attempt < TOTAL_ATTEMPTS:
            raise
        # Last attempt for THIS x
        if final_ok_check(exc, x):
            return f"accepted:{x}"
        raise

@flow
def parent():
    futures = my_task.map(range(10))
    return futures.result(raise_on_failure=False)

So, mapping won’t combine attempts across items — each item’s retries and “final decision” remain scoped to that item.