Great questions! 1) Adding the deployment name to your Slack message In Prefect 3.x automations, the template context includes native objects for the triggering event if you reference them. For a flow-run crash that came from a deployment, you can use

deployment.name

directly. For example, you can simplify your body to:

"Flow {{ flow_run.name }} crashed\n"
"Deployment: {{ deployment.name if deployment else 'N/A' }}\n"
"<" + prefect_ui_url + "/flow-runs/flow-run/{{ event.resource.id | replace('prefect.flow-run.', '') }}|View in Prefect UI>"

If you prefer to stick with the event-only context, your current selector is fine — just add a fallback:

"Deployment: {{ (event.related
  | selectattr('prefect.resource.role', 'equalto', 'deployment')
  | map(attribute='prefect.resource.name')
  | first) or 'N/A' }}"

Notes: -

deployment

is only present for runs created from a deployment; ad-hoc runs won’t have it, so keep the

N/A

fallback. -

prefect.resource.name

is the display name you want for notifications. 2) “Retry a crashed flow X times” - Crashed is a terminal state in Prefect; flows themselves do not have retries. Only tasks support retries (via

@task(retries=..., retry_delay_seconds=...)

). - There isn’t a built-in “retry this same flow run X times” feature. - If you want to automatically try again after a crash, the common pattern is to use an Automation that listens for

prefect.flow-run.Crashed

and then creates a new run of the same deployment (i.e., “re-run” the deployment). Ways to control re-runs: - Rate limiting: Use

threshold

and

within

on the trigger to only fire after N crashes within a time window. This doesn’t cap “X total attempts,” but it prevents rapid loops. - Hard cap on attempts: There isn’t a built-in “max attempts per original run” cap in automations today. If you truly need a strict X attempts, options are: - Encode an

attempt

parameter in your flow and have your flow exit early when it exceeds X. Then, when you re-run from an automation, pass

attempt = (flow_run.parameters.attempt or 0) + 1

in the action parameters. Be aware Jinja renders strings, so cast in your flow if needed. - Or handle counters externally (e.g., store per-key attempt counts in a DB/Redis and stop re-running when the count is reached). If you want, I can sketch a companion automation that: - Triggers on

prefect.flow-run.Crashed

- Re-runs the same deployment after a small delay - Adds a threshold/within window to avoid rapid loops - Optionally forwards the original parameters and increments an

attempt

parameter To confirm your intent: would you like to re-run the same deployment automatically after a crash, and do you want to limit “X attempts” per unit time (rate limiting) or truly “X attempts per original run”?

Awesome — here’s a second automation that will re-run the same deployment whenever a flow run crashes. Notes: - This uses RunDeployment with

source="inferred"

so it will re-run the deployment associated with the crashed flow run. - I added a small delay to avoid immediate thrash. You can adjust or remove

schedule_after

. - Optional: add `threshold`/`within` if you want to avoid rapid loops (e.g., only re-run after N crashes in a time window).

from datetime import timedelta
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.actions import RunDeployment

rerun_on_crash_automation = Automation(
    name="Re-run deployment on crash",
    description="Automatically re-run the same deployment after a crash",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Crashed"},
        match={"prefect.resource.id": "prefect.flow-run.*"},
        posture=Posture.Reactive,
        # Optional rate-limiting:
        # threshold=2,                      # require 2 crashes
        # within=timedelta(minutes=5),      # within 5 minutes
    ),
    actions=[
        RunDeployment(
            source="inferred",                 # re-run the same deployment
            schedule_after=timedelta(seconds=30),  # small delay before re-running
            # Optional: forward parameters from the original run (if desired)
            # parameters="{{ flow_run.parameters }}"
        )
    ],
)

# Persist the automation
rerun_on_crash_automation.create()

If you’d prefer to only re-run a specific deployment, switch to

source="selected"

and pass

deployment_id=...

. Want me to add a strict “max X attempts” pattern too (e.g., by passing/incrementing an

attempt

parameter and stopping after a limit)?

Great — here’s a pattern that re-runs the same deployment after a crash but caps retries at X attempts using parameters and a guard at the start of your flow. 1) Automation to re-run on crash and increment attempts - Uses

source="inferred"

so it re-runs the same deployment that crashed. - Increments an

attempt

parameter and carries a

max_attempts

value. - Adds a small delay to avoid thrashing. - Optional: add `threshold`/`within` if you want rate-limiting.

from datetime import timedelta
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.actions import RunDeployment

MAX_ATTEMPTS = 3  # set your limit

rerun_on_crash_with_cap = Automation(
    name="Re-run deployment on crash (max attempts)",
    description=f"Re-run the same deployment after a crash up to {MAX_ATTEMPTS} attempts",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Crashed"},
        match={"prefect.resource.id": "prefect.flow-run.*"},
        posture=Posture.Reactive,
        # Optional rate-limiting (uncomment as needed):
        # threshold=2,
        # within=timedelta(minutes=5),
    ),
    actions=[
        RunDeployment(
            source="inferred",
            schedule_after=timedelta(seconds=30),
            parameters={
                # Jinja rendering: incoming values will be strings at runtime;
                # your flow should cast to int.
                "attempt": "{{ (flow_run.parameters.attempt | default(0) | int) + 1 }}",
                "max_attempts": MAX_ATTEMPTS,
                "original_run_id": "{{ flow_run.id }}",
            },
        )
    ],
)

rerun_on_crash_with_cap.create()

2) Add a guard at the top of your flow - Cast `attempt`/`max_attempts` to integers because Jinja renders strings. - Exit early (successfully) when attempts exceed the cap. This prevents further crash events from triggering the automation again.

from prefect import flow, get_run_logger

@flow
def my_flow(attempt: int = 0, max_attempts: int = 3, original_run_id: str | None = None):
    logger = get_run_logger()
    # Jinja renders strings — cast to int
    attempt = int(attempt or 0)
    max_attempts = int(max_attempts or 0)

    if attempt > max_attempts:
        logger.warning(
            f"Max attempts reached ({attempt - 1}/{max_attempts}). "
            "Not retrying further."
        )
        return

    # Your flow logic below
    <http://logger.info|logger.info>(f"Attempt {attempt}/{max_attempts} (original_run_id={original_run_id})")
    # ...

Notes and caveats: - This caps retries per “chain” using parameters. It’s robust as long as the new run reaches the guard. If your infrastructure crashes before the flow starts running, the guard can’t execute; consider adding `threshold`/`within` to reduce any potential storm. - If you want to surface attempt info in your Slack message, you can include

{{ flow_run.parameters.attempt }}

and/or

{{ flow_run.parameters.max_attempts }}

in your notification template. - If you prefer a single deployment (instead of inferred), switch to

source="selected"

and provide

deployment_id=...

.

Got it — you want the new run to use the exact same parameters as the run that crashed. In Prefect 3.x, there isn’t a built-in “rerun with same parameters” button, but there are two clean ways to do this: A) Automations: trigger a new deployment run and pass original parameters - Create an Automation that fires on flow-run failure and uses a RunDeployment action. - In the action’s parameters, use Jinja to reference the original flow run’s parameters. Example parameter mapping in the action (UI or Python):

{
  "my_param": {
    "__prefect_kind": "jinja",
    "template": "{{ flow_run.parameters['my_param'] }}"
  },
  "another_param": {
    "__prefect_kind": "jinja",
    "template": "{{ flow_run.parameters['another_param'] }}"
  }
}

Helpful docs: - Automations concepts - How to access parameters in templates - Python API: actions (RunDeployment) B) Programmatically: create a new run with the same parameters Use the client to read the failed flow run, extract its parameters, and create a new run.

import asyncio
from uuid import UUID
from prefect.client.orchestration import get_client

async def rerun_with_same_params(failed_run_id: str | UUID):
    async with get_client() as client:
        original = await client.read_flow_run(UUID(str(failed_run_id)))

        # Get the flow object for create_flow_run
        flows = await client.read_flows(flow_filter={"id": {"any_": [original.flow_id]}})
        flow = flows[0]

        new_run = await client.create_flow_run(
            flow=flow,
            name=f"{original.name}-retry",
            parameters=original.parameters,
            tags=(original.tags or []) + ["rerun"],
        )

        print(f"New run: {new_run.id}")
        return new_run.id

# Usage
# asyncio.run(rerun_with_same_params("your-failed-run-id"))

Client/CLI references: - Python API: client (read_flow_run, create_flow_run) - CLI: flow-run commands A couple notes: - There’s no native “rerun”/“clone” button in the UI today. - If you prefer to stick with deployments, Automations + RunDeployment is the smoothest “auto-rerun with same params” approach. Quick question to tailor this: do you want to trigger the new run automatically (via Automation), or manually (via a small script/CLI)? If you share the failed run ID and parameter names, I can draft the exact automation or script for you.

Absolutely — here’s a ready-to-use Prefect 3.x Automation that will: - Trigger on a crashed flow run - Re-run the same deployment - Pass along the original parameters - Increment a

retry_count

parameter - Enforce a

max_retries

limit (your flow should check this and avoid crashing when the limit is reached) Notes and constraints: - Automations don’t currently support conditional/guarded actions, so the “stop after X” needs to be enforced in your flow (e.g., exit gracefully when

retry_count >= max_retries

). Since this automation only triggers on Crashed, a non-crashing exit will terminate the retry loop. - This assumes your flow accepts

retry_count

and

max_retries

as parameters (with sensible defaults). If they aren’t present, the automation’s Jinja template will still pass them. - The action uses

source="inferred"

so it re-runs the same deployment that produced the crash event (flows without a deployment won’t be retried). Full example (Python) to create the automation:

from datetime import timedelta
from prefect.automations import (
    Automation,
    EventTrigger,
    Posture,
    RunDeployment,
)

def create_crash_retry_automation(
    name: str = "Auto-retry crashed flow runs with same parameters",
    max_retries: int = 3,
    delay_seconds: int = 0,
):
    """
    Creates an Automation that:
      - fires on flow-run crashes
      - re-runs the same deployment
      - passes all original parameters AND increments `retry_count`
      - includes a fixed `max_retries`

    Requirements:
      - Your flow should accept `retry_count` and `max_retries` parameters.
      - At the start of your flow, exit gracefully (do not crash) when `retry_count >= max_retries`.
        This prevents further retries since the automation only listens for 'Crashed'.
    """

    # This Jinja template copies the original parameters, injects +1 retry_count, and max_retries.
    # It constructs the entire parameter dict in one go.
    params_template = """
{% set p = flow_run.parameters.copy() if flow_run and flow_run.parameters else {} %}
{% set _ = p.update({
  'retry_count': (p.get('retry_count', 0) | int) + 1,
  'max_retries': MAX_RETRIES_PLACEHOLDER
}) %}
{{ p | tojson }}
""".strip().replace("MAX_RETRIES_PLACEHOLDER", str(int(max_retries)))

    automation = Automation(
        name=name,
        trigger=EventTrigger(
            expect={"prefect.flow-run.Crashed"},
            # Tip: add a ResourceSpecification to limit to a specific deployment or flow if desired
            # match_related=ResourceSpecification({
            #     "prefect.resource.role": "deployment",
            #     "prefect.resource.name": "your-deployment-name"
            # }),
            posture=Posture.Reactive,
            threshold=1,
            within=timedelta(seconds=0),
        ),
        actions=[
            RunDeployment(
                # "inferred" will target the same deployment tied to the crashed flow run
                source="inferred",
                # Optional: delay the retry a bit to avoid immediate hammering
                schedule_after=timedelta(seconds=delay_seconds),
                parameters={
                    # Rebuild the entire parameter mapping with original params + retry_count/max_retries
                    "__prefect_kind": "json",
                    "value": {
                        "__prefect_kind": "jinja",
                        "template": params_template,
                    },
                },
            )
        ],
    )

    # Creates the automation in your workspace (returns the saved Automation)
    return automation.create()

Example flow pattern you can use to enforce the retry limit: ``` from prefect import flow, get_run_logger @flow def my_flow(..., retry_count: int = 0, max_retries: int = 3): logger = get_run_logger() if retry_count >= max_retries: logger.info(f"Max retries reached ({retry_count}/{max_retries}); exiting without crash.") return

# Your logic here # If this run crashes (process or worker crash), the automation will retry. # If you raise an exception (Failed), the automation won't trigger unless you also listen for Failed. ``` Helpful references: - Automations concepts: docs - Using Jinja to access flow run parameters in actions: guide Want me to: - Restrict this to one specific deployment or flow? - Add a small backoff (e.g., exponential) using

schedule_after

? - Emit a notification after the final attempt?