In Prefect 3.x, there isn't a built-in mechanism to automatically resume a flow from the point of failure while skipping successful tasks. However, you have a few options to achieve this: 1. Use Task-Level Caching

python
from prefect import flow, task
from prefect.tasks import task_input_hash

@task(cache_key_fn=task_input_hash)
def expensive_task(x):
    # This task will only run if the inputs change
    # or if the cache expires
    return x * 2

@flow
def my_flow(x):
    result = expensive_task(x)
    # ... rest of your flow

2. Implement Custom State Checking

python
from prefect import flow, task
from prefect.context import get_run_context

@task
def check_previous_state(task_name):
    context = get_run_context()
    # You can implement logic here to check if the task
    # was successful in a previous run
    return should_skip

@flow
def my_flow():
    if not check_previous_state("task_1"):
        task_1_result = task_1()
    if not check_previous_state("task_2"):
        task_2_result = task_2()

3. Use Flow-Level Controls You can use Prefect's flow control mechanisms to implement more sophisticated retry logic:

python
from prefect import flow
from prefect.flow_runs import resume_flow_run

@flow
def my_flow():
    try:
        # Your flow logic here
        pass
    except Exception as e:
        # Capture state and handle resumption
        current_state = capture_current_state()
        resume_flow_run(flow_run_id=get_run_context().flow_run.id, 
                       run_input={"previous_state": current_state})

A few important points to consider: 1. Task caching is the most straightforward approach if you want to skip previously successful tasks. Tasks with the same inputs will use cached results instead of re-executing. 2. For more complex scenarios, you'll need to implement your own state tracking mechanism. This could involve storing task states in a database or using Prefect's state management capabilities. 3. The solution you implement should be tailored to your specific use case: - For simple workflows, task caching might be sufficient - For complex workflows, you might need a combination of caching and custom state management - For critical workflows, consider implementing proper error handling and state persistence Remember that while Prefect doesn't provide automatic task skipping on resume, it gives you the building blocks to implement this behavior in a way that best suits your needs. Would you like me to elaborate on any of these approaches or provide more specific examples for your use case?