@Marvin I need to disable running late tasks. is possible to mark Late tasks as failed so they don't run again when server/agent is online, or there is another way to do it?

It seems there isn't a straightforward way in Prefect to mark late tasks as failed automatically. However, you can potentially control task executions using workarounds with your workflow design. Here are a couple of options: 1. Constrain Task Concurrency to Zero: By setting the concurrency limit for specific tasks to zero, you can prevent these tasks from running altogether. When the concurrency limit is set to zero, any attempt to run the task will be aborted. 2. Custom Conditional Logic in Tasks: You can include conditional checks within your task functions to determine if they should proceed based on the current time compared to the scheduled time, and raise an intentional failure if they are late. Example for Concurrency Limit:

python
from prefect import task, Flow
from prefect.tasks.control_flow import merge_task

@task
def check_execution_window():
    import datetime
    now = datetime.datetime.utcnow()
    # Assuming tasks should only run between 9 AM - 5 PM UTC
    if now.hour < 9 or now.hour > 17:
        raise ValueError("Task execution window has passed")

with Flow("conditional_task_execution") as flow:
    execution_check = check_execution_window()
    # downstream tasks here that depend on execution_check

Example for Custom Conditional Logic within Task:

python
from prefect import task, Flow

@task
def my_task():
    import datetime
    now = datetime.datetime.utcnow()
    scheduled_time = datetime.datetime(2023, 10, 1, 10)  # Example scheduled time
    if (now - scheduled_time).total_seconds() > 3600:  # 1 hour late
        raise ValueError("Task is too late to run")

with Flow("custom_logic_task") as flow:
    my_task_run = my_task()

Neither of these methods fully automates marking late tasks as failed, but they allow you to control whether tasks run based on time conditions. For more advanced uses or other ideas, you can check out the Prefect documentation on tasks and concurrency handling here. If you require a more automated feature, you might want to raise a feature request on GitHub: Prefect GitHub Issues.