Sorry, I encountered an error while processing your request. Please try again.

You can scope the automation to specific work pools by adding a match_related filter for the work-pool related resource. Flow-run heartbeat events include the work pool as a related resource (role:

work-pool

), so use

match_related

to target only those pools. Here’s your automation updated to only apply to a given list of work pools:

from datetime import timedelta

from prefect.automations import Automation
from prefect.client.schemas.objects import StateType
from prefect.events.actions import ChangeFlowRunState
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.schemas.events import ResourceSpecification

TARGET_POOLS = ["pool-a", "pool-b"]  # replace with your work pool names

my_automation = Automation(
    name="Crash zombie flows (selected pools)",
    trigger=EventTrigger(
        after={"prefect.flow-run.heartbeat"},
        expect={
            "prefect.flow-run.heartbeat",
            "prefect.flow-run.Completed",
            "prefect.flow-run.Failed",
            "prefect.flow-run.Cancelled",
            "prefect.flow-run.Crashed",
        },
        # primary resource: flow runs
        match=ResourceSpecification({"prefect.resource.id": ["prefect.flow-run.*"]}),
        # scope to related work pools by name
        match_related=ResourceSpecification({
            "prefect.resource.role": ["work-pool"],
            "prefect.resource.name": TARGET_POOLS,
        }),
        for_each={"prefect.resource.id"},
        posture=Posture.Proactive,
        threshold=1,
        within=timedelta(seconds=90),
    ),
    actions=[
        ChangeFlowRunState(
            state=StateType.CRASHED,
            message="Flow run marked as crashed due to missing heartbeats.",
        )
    ],
)

if __name__ == "__main__":
    my_automation.create()

Notes: -

match_related

is required here because the work pool is attached to the flow-run event as a related resource, not the primary resource. - Use

prefect.resource.name

to match by work pool name and include multiple pools by passing a list. - Keeping

prefect.resource.role: work-pool

ensures you’re matching the correct related resource type. - Runs that aren’t associated with a work pool (e.g., ad-hoc/local runs) will be excluded by this filter.

Sorry, I encountered an error while processing your request. Please try again.

Use a single match_related spec and pass a list of pool names for prefect.resource.name. List values are treated as OR within a single ResourceSpecification. Example:

from datetime import timedelta

from prefect.automations import Automation
from prefect.client.schemas.objects import StateType
from prefect.events.actions import ChangeFlowRunState
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.schemas.events import ResourceSpecification

TARGET_POOLS = ["pool-a", "pool-b", "pool-c"]  # any of these pools

my_automation = Automation(
    name="Crash zombie flows (selected pools)",
    trigger=EventTrigger(
        after={"prefect.flow-run.heartbeat"},
        expect={
            "prefect.flow-run.heartbeat",
            "prefect.flow-run.Completed",
            "prefect.flow-run.Failed",
            "prefect.flow-run.Cancelled",
            "prefect.flow-run.Crashed",
        },
        match=ResourceSpecification({"prefect.resource.id": ["prefect.flow-run.*"]}),
        match_related=ResourceSpecification({
            "prefect.resource.role": "work-pool",
            "prefect.resource.name": TARGET_POOLS,  # OR across these names
        }),
        for_each={"prefect.resource.id"},
        posture=Posture.Proactive,
        threshold=1,
        within=timedelta(seconds=90),
    ),
    actions=[
        ChangeFlowRunState(
            state=StateType.CRASHED,
            message="Flow run marked as crashed due to missing heartbeats.",
        )
    ],
)

if __name__ == "__main__":
    my_automation.create()

Notes: - Using a list for

prefect.resource.name

performs an OR match across those pool names. - Keep

prefect.resource.role: work-pool

to ensure you’re matching the correct related resource. - Avoid using a list of ResourceSpecifications for this case—multiple specs are ANDed together, not ORed.