Yes — this is a common pattern, and the main thing to know is:

persist_result=True

by itself does not mean “reuse successful task results on retry” unless you are also using task caching. So the recommendation is usually: 1. Use task caching for tasks you want to skip when inputs are unchanged 2. Add a runtime switch that can force those tasks to ignore cache and rerun A good pattern looks like this:

python
from prefect import flow, task
from prefect.cache_policies import INPUTS

@task(
    persist_result=True,
    cache_policy=INPUTS,
)
def task_a(x: int) -> int:
    print("running task_a")
    return x + 1

@task(
    persist_result=True,
    cache_policy=INPUTS,
)
def task_b(y: int) -> int:
    print("running task_b")
    return y * 2

@flow
def my_flow(x: int, force_rerun: bool = False):
    a = task_a.with_options(refresh_cache=force_rerun)(x)
    b = task_b.with_options(refresh_cache=force_rerun)(a)
    return b

How this behaves: -

force_rerun=False

- successful tasks can reuse cached/persisted results -

force_rerun=True

- tasks will rerun even if a cached result already exists So for your use case, I’d recommend: Normal case - user runs / retries flow normally - successful tasks are reused Special case - expose a flow parameter like

force_rerun=True

- when user wants a “clean” retry, they start a new run with that parameter enabled Important limitation with UI retry If the user clicks Retry on an existing flow run in the UI, that retry generally reuses the same parameters as the original run. That means you usually cannot change

force_rerun

during that same retry action. Because of that, the most practical UX is: - for ordinary retries: use UI Retry - for “rerun everything even if cached”: create a new flow run with

force_rerun=True

If you want partial control, you can do it per task:

python
@flow
def my_flow(x: int, rerun_task_a: bool = False, rerun_task_b: bool = False):
    a = task_a.with_options(refresh_cache=rerun_task_a)(x)
    b = task_b.with_options(refresh_cache=rerun_task_b)(a)
    return b

That lets users decide which successful tasks to recompute. A few notes: -

persist_result=True

stores results -

cache_policy=INPUTS

tells Prefect when a previous result is reusable -

refresh_cache=True

forces recomputation even if cache exists So the core idea is: Use

cache_policy=INPUTS

+

persist_result=True

+

.with_options(refresh_cache=...)

Relevant docs: - Caching - Write tasks - Results If you want, I can also show you: 1. a full example wired to deployment parameters, or 2. a pattern where UI retry automatically reruns everything after a failure marker / flag.

Yes, via SDK/API: probably yes. No, via the standard UI retry button: no. The important distinction is: - UI manual retry retries the existing flow run with its current stored parameters - there is no UI step to edit parameters before clicking Retry - but with the Python SDK / REST API you can first update the flow run’s

parameters

, then set its state to retry So if your question is: “Can I modify an existing failed flow run so that when it is retried it uses

force_rerun=True

?” The answer is: yes, using

update_flow_run(...)

before retrying it programmatically. Example with Python SDK:

python
from uuid import UUID
from prefect import get_client
from prefect.states import Scheduled

async def retry_with_force_rerun(flow_run_id: str):
    async with get_client() as client:
        fr_id = UUID(flow_run_id)

        flow_run = await client.read_flow_run(fr_id)

        new_params = {
            **(flow_run.parameters or {}),
            "force_rerun": True,
        }

        await client.update_flow_run(
            flow_run_id=fr_id,
            parameters=new_params,
        )

        await client.set_flow_run_state(
            flow_run_id=fr_id,
            state=Scheduled(message="Retry with force_rerun=True"),
            force=True,
        )

That should give you the behavior you want if your flow reads that parameter and applies it to tasks, e.g.:

python
@flow
def my_flow(force_rerun: bool = False):
    a = task_a.with_options(refresh_cache=force_rerun)()
    b = task_b.with_options(refresh_cache=force_rerun)()

A couple of caveats: - This is not the same as clicking Retry in the UI afterward. If you update the parameters and then use the UI Retry button later, the run should retry with whatever parameters are stored at that point, but there is no official “edit before retry” UI workflow. - Reusing the same flow run with changed parameters is a bit unusual operationally. In many teams, it’s cleaner to create a new flow run with

force_rerun=True

instead of mutating an old failed one. CLI note: -

prefect flow-run retry

exists, but it does not provide a way to override parameters inline. - Verified command:

prefect flow-run retry --help

Docs / references: - Retry flow runs - Python client reference - Caching My practical recommendation: - Best UX: add

force_rerun

as a deployment/run parameter and have users start a new run when they want to ignore cache - If you must retry the same run: build a small internal script/API endpoint that: 1. reads the failed flow run 2. updates parameters with

force_rerun=True

3. sets the run back to

Scheduled

If you want, I can give you a complete production-ready helper for: - retry same flow run with updated params, or - clone failed flow run into a new run with

force_rerun=True

.

Sure — below is a production-ready helper for retrying the same flow run with updated parameters in Prefect 3.x. It does all of this: - validates the target flow run exists - merges existing parameters with your overrides - optionally checks the current state - updates the flow run parameters - retries the same flow run by setting it to

Scheduled

- returns a structured result - logs useful messages - raises clear exceptions on failure Important note: this approach is supported by the SDK, but it is still a somewhat advanced workflow. If your team wants a simpler audit trail, creating a new flow run is often cleaner. But if you explicitly want to reuse the same flow run, this helper is the right pattern. Code: ```python from future import annotations from dataclasses import dataclass from typing import Any, Mapping from uuid import UUID from prefect import get_client from prefect.client.schemas.objects import StateType from prefect.states import Scheduled class FlowRunRetryError(Exception): """Base error for flow run retry helper.""" class FlowRunNotFoundError(FlowRunRetryError): """Raised when the flow run does not exist.""" class FlowRunStateError(FlowRunRetryError): """Raised when the flow run is in a state that should not be retried.""" class FlowRunUpdateError(FlowRunRetryError): """Raised when parameters could not be updated.""" class FlowRunRetryRejectedError(FlowRunRetryError): """Raised when orchestration rejects the retry state transition.""" @dataclass class RetryFlowRunResult: flow_run_id: UUID flow_run_name: str | None old_parameters: dict[str, Any] new_parameters: dict[str, Any] old_state_type: str | None old_state_name: str | None orchestration_status: str orchestration_details: str | None async def retry_same_flow_run_with_updated_params( flow_run_id: str | UUID, parameter_overrides: Mapping[str, Any], *, require_terminal_state: bool = True, allow_states: set[str] | None = None, message: str | None = None, ) -> RetryFlowRunResult: """ Update an existing Prefect flow run's parameters, then retry the same flow run. Args: flow_run_id: The ID of the existing flow run. parameter_overrides: Parameters to merge into the flow run's existing parameters. Example: {"force_rerun": True} require_terminal_state: If True, only allow retry when the current state is terminal (FAILED, CRASHED, COMPLETED, CANCELLED). allow_states: Optional explicit allowlist of state type names, e.g. {"FAILED", "CRASHED", "COMPLETED"}. If provided, this takes precedence over require_terminal_state. message: Optional message attached to the Scheduled retry state. Returns: RetryFlowRunResult with details of the update + retry attempt. Raises: FlowRunNotFoundError FlowRunStateError FlowRunUpdateError FlowRunRetryRejectedError """ run_id = UUID(str(flow_run_id)) async with get_client() as client: try: flow_run = await client.read_flow_run(run_id) except Exception as exc: raise FlowRunNotFoundError(f"Could not read flow run {run_id}") from exc if flow_run is None: raise FlowRunNotFoundError(f"Flow run {run_id} was not found") current_state = flow_run.state current_state_type = current_state.type if current_state else None current_state_name = current_state.name if current_state else None current_state_type_name = ( current_state_type.name if hasattr(current_state_type, "name") else None ) if allow_states is not None: normalized = {s.upper() for s in allow_states} if current_state_type_name not in normalized: raise FlowRunStateError(

f"Flow run {run_id} is in state {current_state_type_name!r}; " f"allowed states are {sorted(normalized)}" ) elif require_terminal_state: terminal_states = { StateType.COMPLETED.name, StateType.FAILED.name, StateType.CRASHED.name, StateType.CANCELLED.name, } if current_state_type_name not in terminal_states: raise FlowRunStateError( f"Flow run {run_id} is in state {current_state_type_name!r}; " f"expected a terminal state in {sorted(terminal_states)}" ) old_parameters = dict(flow_run.parameters or {}) new_parameters = {*old_parameters, *dict(parameter_overrides)} try: await client.update_flow_run( flow_run_id=run_id, parameters=new_parameters, ) except Exception as exc: raise FlowRunUpdateError( f"Failed to update parameters for flow run {run_id}" ) from exc retry_message = message or "Retrying flow run with updated parameters" try: state_result = await client.set_flow_run_state( flow_run_id=run_id, state=Scheduled(message=retry_message), force=True, ) except Exception as exc: raise FlowRunRetryRejectedError( f"Failed to set flow run {run_id} to Scheduled for retry" ) from exc status_name = ( state_result.status.name if getattr(state_result, "status", None) is not None else "UNKNOWN" ) details_reason = None if getattr(state_result, "details", None) is not None: details_reason = getattr(state_result.details, "reason", None) if status_name not in {"ACCEPT", "ABORT", "WAIT", "REJECT"}: # Defensive: handle unexpected statuses explicitly details_reason = details_reason or "Unknown orchestration response" if status_name != "ACCEPT": raise FlowRunRetryRejectedError( f"Retry orchestration was not accepted for flow run {run_id}. " f"Status={status_name}. Details={details_reason}" ) return RetryFlowRunResult( flow_run_id=run_id, flow_run_name=getattr(flow_run, "name", None), old_parameters=old_parameters, new_parameters=new_parameters, old_state_type=current_state_type_name, old_state_name=current_state_name, orchestration_status=status_name, orchestration_details=details_reason, )

Minimal usage:

```python
import asyncio

async def main():
    result = await retry_same_flow_run_with_updated_params(
        flow_run_id="YOUR-FLOW-RUN-ID",
        parameter_overrides={"force_rerun": True},
    )
    print(result)

asyncio.run(main())

Recommended usage in your case:

python
import asyncio

async def main():
    result = await retry_same_flow_run_with_updated_params(
        flow_run_id="YOUR-FLOW-RUN-ID",
        parameter_overrides={
            "force_rerun": True,
        },
        allow_states={"FAILED", "CRASHED", "COMPLETED"},
        message="Manual retry with force_rerun=True",
    )

    print("Retried flow run:", result.flow_run_id)
    print("Old params:", result.old_parameters)
    print("New params:", result.new_parameters)

asyncio.run(main())

And your flow/tasks should be wired like this: ```python from prefect import flow, task from prefect.cache_policies import INPUTS @task(persist_result=True, cache_policy=INPUTS) def task_a(x: int) -> int: print("running task_a") return x + 1 @task(persist_result=True, cache_policy=INPUTS) def task_b(y: int) -> int: print("running task_b") return y * 2 @flow

def my_flow(x: int, force_rerun: bool = False) -> int: a = task_a.with_options(refresh_cache=force_rerun)(x) b = task_b.with_options(refresh_cache=force_rerun)(a) return b ``` If you want a slightly more operational version, here is a helper with logging: ```python from future import annotations import logging from dataclasses import asdict, dataclass from typing import Any, Mapping from uuid import UUID from prefect import get_client from prefect.client.schemas.objects import StateType from prefect.states import Scheduled logger = logging.getLogger(name) class FlowRunRetryError(Exception): pass class FlowRunNotFoundError(FlowRunRetryError): pass class FlowRunStateError(FlowRunRetryError): pass class FlowRunUpdateError(FlowRunRetryError): pass class FlowRunRetryRejectedError(FlowRunRetryError): pass @dataclass class RetryFlowRunResult: flow_run_id: UUID flow_run_name: str | None old_parameters: dict[str, Any] new_parameters: dict[str, Any] old_state_type: str | None old_state_name: str | None orchestration_status: str orchestration_details: str | None def asdict(self) -> dict[str, Any]: return asdict(self) async def retry_same_flow_run_with_updated_params( flow_run_id: str | UUID, parameter_overrides: Mapping[str, Any], *, require_terminal_state: bool = True, allow_states: set[str] | None = None, message: str | None = None, ) -> RetryFlowRunResult: run_id = UUID(str(flow_run_id)) logger.info("Preparing retry for flow run %s", run_id) async with get_client() as client: try: flow_run = await client.read_flow_run(run_id) except Exception as exc: logger.exception("Unable to read flow run %s", run_id) raise FlowRunNotFoundError(f"Could not read flow run {run_id}") from exc if flow_run is None: raise FlowRunNotFoundError(f"Flow run {run_id} was not found") current_state = flow_run.state current_state_type = current_state.type if current_state else None current_state_name = current_state.name if current_state else None current_state_type_name = ( current_state_type.name if hasattr(current_state_type, "name") else None ) logger.info( "Current state for flow run %s: type=%s name=%s", run_id, current_state_type_name, current_state_name, ) if allow_states is not None: normalized = {s.upper() for s in allow_states} if current_state_type_name not in normalized: raise FlowRunStateError( f"Flow run {run_id} is in state {current_state_type_name!r}; " f"allowed states are {sorted(normalized)}" ) elif require_terminal_state: terminal_states = { StateType.COMPLETED.name, StateType.FAILED.name, StateType.CRASHED.name, StateType.CANCELLED.name, } if current_state_type_name not in terminal_states: raise FlowRunStateError( f"Flow run {run_id} is in state {current_state_type_name!r}; " f"expected terminal state in {sorted(terminal_states)}" ) old_parameters = dict(flow_run.parameters or {}) new_parameters = {*old_parameters, *dict(parameter_overrides)} logger.info( "Updating parameters for flow run %s. Overrides=%s", run_id, dict(parameter_overrides), ) try: await client.update_flow_run( flow_run_id=run_id, parameters=new_parameters, ) except Exception as exc: logger.exception("Parameter update failed for flow run %s", run_id) raise FlowRunUpdateError(

f"Failed to update parameters for flow run {run_id}" ) from exc retry_message = message or "Retrying flow run with updated parameters" logger.info( "Submitting retry for flow run %s with message=%r", run_id, retry_message, ) try: state_result = await client.set_flow_run_state( flow_run_id=run_id, state=Scheduled(message=retry_message), force=True, ) except Exception as exc: logger.exception("Retry scheduling failed for flow run %s", run_id) raise FlowRunRetryRejectedError( f"Failed to set flow run {run_id} to Scheduled for retry" ) from exc status_name = ( state_result.status.name if getattr(state_result, "status", None) is not None else "UNKNOWN" ) details_reason = ( getattr(state_result.details, "reason", None) if getattr(state_result, "details", None) is not None else None ) logger.info( "Orchestration response for flow run %s: status=%s details=%s", run_id, status_name, details_reason, ) if status_name != "ACCEPT": raise FlowRunRetryRejectedError( f"Retry orchestration was not accepted for flow run {run_id}. " f"Status={status_name}. Details={details_reason}" ) result = RetryFlowRunResult( flow_run_id=run_id, flow_run_name=getattr(flow_run, "name", None), old_parameters=old_parameters, new_parameters=new_parameters, old_state_type=current_state_type_name, old_state_name=current_state_name, orchestration_status=status_name, orchestration_details=details_reason, ) logger.info("Retry submitted successfully: %s", result.asdict()) return result

If you want a tiny wrapper specifically for your `force_rerun=True` case:

```python
async def force_rerun_existing_flow_run(flow_run_id: str | UUID) -> RetryFlowRunResult:
    return await retry_same_flow_run_with_updated_params(
        flow_run_id=flow_run_id,
        parameter_overrides={"force_rerun": True},
        allow_states={"FAILED", "CRASHED", "COMPLETED"},
        message="Manual retry with force_rerun=True",
    )

Production notes / recommendations 1. Be careful with COMPLETED runs Retrying a completed run is possible in some cases, but operationally it can be surprising. If you want stricter behavior, use:

python
   allow_states={"FAILED", "CRASHED"}

2. Parameter schema compatibility If your deployment/flow expects typed parameters, ensure the override values are JSON-serializable and valid for the flow signature. 3. Cached tasks still need correct wiring This helper only changes the flow run parameter. Your flow code must actually use

force_rerun

to set

refresh_cache=True

. 4. Auditability Reusing the same flow run changes the historical run’s parameters before retry. If audit clarity matters, create a new run instead. Useful references: - Retry flow runs - Caching - Prefect client reference If you want, I can also prepare: - a sync version of this helper - a small CLI script like

python retry_flow.py --flow-run-id ... --force-rerun

- a FastAPI endpoint for internal ops use

Correct — you cannot create a second new run for the same flow with the same

idempotency_key

. A couple of precise points: - Prefect uses the term

idempotency_key

- Idempotency is scoped to

(flow_id, idempotency_key)

- If you call “create flow run” again with the same flow and same key, Prefect will return the existing run, not create a new one So your assumption is right. That means if your external app uses

idempotency_key

for exactly-once semantics, then: - same key => same existing flow run - new key => new flow run This creates a design tension with your “force rerun as a new audited run” requirement: - from the external system’s perspective, exactly-once says don’t create another run for the same business event - from the operator’s perspective, force rerun says I want a second execution attempt as a distinct run Those are both valid, but they mean you should separate business event identity from execution attempt identity. Recommended design Use two identifiers: 1. business id — stable across retries/reruns Example: order id, message id, job id 2. idempotency key — unique per execution attempt For example: -

business_id = invoice-12345

- first run: -

idempotency_key = invoice-12345:attempt-1

- forced rerun: -

idempotency_key = invoice-12345:attempt-2

Then pass

business_id

as a normal flow parameter, tag, or custom metadata, so you can still correlate all attempts. Example payload idea:

python
parameters = {
    "business_id": "invoice-12345",
    "force_rerun": True,
}
idempotency_key = "invoice-12345:attempt-2"

That gives you: - exactly once per attempt submission - multiple auditable attempts for the same business event - easy correlation in your app and Prefect What not to do Do not use the raw business event id itself as the permanent idempotency key if you ever want a true second run later. Bad for your use case:

python
idempotency_key = "invoice-12345"

Because then every later “create run” call for that event just returns the original run forever. Better patterns Pattern A: attempt counter in your app - your app stores attempt number - creates: -

invoice-12345:attempt-1

-

invoice-12345:attempt-2

Pattern B: operator-triggered rerun key - first automatic submission: -

invoice-12345

- manual rerun: -

invoice-12345:manual-rerun:2026-05-11T12:00:00Z

This works technically, but semantically it means the first key is “event identity” and later keys are “attempt identity”, which is a little inconsistent. I’d prefer Pattern A. Pattern C: UUID per attempt + business id as parameter -

idempotency_key = <new uuid>

-

parameters["business_id"] = "invoice-12345"

This is the cleanest technically if your app can track attempts elsewhere. Important implication If you need: - exactly one run per upstream request retry - but also the ability to intentionally launch a fresh run later then the external app should generate: - the same

idempotency_key

for retried delivery of the same attempt - a different

idempotency_key

for an intentional new attempt That is usually the correct architecture. So in short: Yes, you are right: You cannot create another new run with the same

idempotency_key

. To create a new auditable rerun, you must use a new idempotency key. References: - FlowRunCreate schema - Create flow run API If you want, I can sketch a recommended idempotency-key strategy for your exact scenario, e.g.: - upstream app retries network calls safely - operators can manually trigger reruns - all attempts remain correlated under one business id.