Yes — in Prefect 3.x you can detect when a run is waiting on a concurrency slot and send a Slack alert if it exceeds a threshold like 60 minutes. There are two common cases: - Flow runs waiting on a deployment-level concurrency limit - Task runs waiting on tag-based concurrency limits In both cases, the run will sit in the AwaitingConcurrencySlot state (a SCHEDULED state) until a slot opens. You can alert on this in two ways: Option A — Use an Automation (recommended) - Create an Automation with a proactive Event trigger: - Resource: Flow runs (for deployment concurrency) or Task runs (for tag-based concurrency) - After: the run enters AwaitingConcurrencySlot - Expect: a subsequent state change (e.g., transition to Running or any other state) - Posture: Proactive - Within: 3600 seconds (60 minutes) - Actions: SendNotification using a SlackWebhook block - This fires when a run remains in AwaitingConcurrencySlot longer than 60 minutes. Notes: - This is very similar to the “zombie flow” detection pattern — same proactive “absence of event within a window” approach, just start from AwaitingConcurrencySlot instead of Running. - You can set it up in the UI under Automations; choose a proactive Event trigger, filter for the run resource, and configure the within window and Slack action. Helpful docs: - Create automations - Event triggers - Detect “zombie” flows (pattern you can adapt) - Tag-based concurrency limits Option B — A small monitoring flow that polls and notifies Slack If you prefer code, here’s a minimal async flow that checks for runs stuck in AwaitingConcurrencySlot and sends Slack notifications after 60 minutes: ``` from datetime import datetime, timedelta, timezone from prefect import flow from prefect.blocks.notifications import SlackWebhook from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, TaskRunFilter, TaskRunFilterState, TaskRunFilterStateName, ) THRESHOLD = timedelta(minutes=60) @flow async def alert_on_long_waits( slack_block_name: str = "my-slack-webhook", check_flow_runs: bool = True, # set False if you only care about task-level tag concurrency check_task_runs: bool = True, # set False if you only care about deployment-level flow concurrency ): slack = await SlackWebhook.load(slack_block_name) now = datetime.now(timezone.utc) async with get_client() as client: if check_flow_runs: frs = await client.read_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["AwaitingConcurrencySlot"]) ) ) ) for fr in frs: if fr.state and fr.state.timestamp and (now - fr.state.timestamp) > THRESHOLD: await slack.notify( body=( f"Flow run '{fr.name}' is waiting on a concurrency slot for " f"{(now - fr.state.timestamp)} (deployment concurrency). " f"Flow run ID: {fr.id}" ) ) if check_task_runs: trs = await client.read_task_runs( task_run_filter=TaskRunFilter( state=TaskRunFilterState( name=TaskRunFilterStateName(any_=["AwaitingConcurrencySlot"]) ) ) ) for tr in trs: if tr.state and tr.state.timestamp and (now - tr.state.timestamp) > THRESHOLD:

await slack.notify( body=( f"Task run '{tr.name}' is waiting on a concurrency slot for " f"{(now - tr.state.timestamp)} (tag-based concurrency). " f"Task run ID: {tr.id} | Flow run ID: {tr.flow_run_id}" ) ) # Schedule this flow to run every few minutes via a deployment. ``` A few tips - Deployment-level concurrency applies to flow runs; tag-based concurrency applies to task runs. Both use the AwaitingConcurrencySlot state while waiting. - If you only care about one of those, set the corresponding check boolean to False in the flow above. - For Automations, you can optionally narrow scope (e.g., only certain deployments or only tasks with specific tags) using match criteria in the event trigger. Questions for you - Do you want to monitor flow runs (deployment concurrency), task runs (tag-based concurrency), or both? - Do you prefer setting this up with an Automation in the UI, or do you want to run a small monitoring flow on a schedule? If you share that, I can give you the exact Automation configuration (or a ready-to-deploy monitoring flow) tailored to your setup.

Short answer: not exactly as written. In Prefect v2.20.16: - Tag-based concurrency limits apply to task runs (not flow runs), and there is no dedicated AwaitingConcurrencySlot state. When a task is blocked by a tag concurrency limit, it remains in Pending and the client retries acquiring a slot every ~30s (configurable via PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS). - Flow runs do not have per-deployment concurrency limits in 2.x; flow-level concurrency is typically governed by your work pool/queue concurrency or infrastructure constraints. What you can do in 2.20.16 - For tasks blocked by tag concurrency: alert on task runs that stay in Pending for > 60 minutes. To reduce noise, filter to tasks with the specific tag(s) that you’ve put under a concurrency limit. Optionally check the state message for hints about concurrency. - For flow runs “stuck” due to work pool/queue limits: alert on flow runs that stay in Scheduled (or Pending, depending on your flow lifecycle) for > 60 minutes. Option A — Monitoring flow (works in OSS and Cloud 2.x) Schedule this flow to run every few minutes; it finds “long-waiting” runs and posts to Slack.

from datetime import datetime, timedelta, timezone

from prefect import flow
from prefect.blocks.notifications import SlackWebhook
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    TaskRunFilter,
    TaskRunFilterState,
    TaskRunFilterStateName,
    TaskRunFilterTags,
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateName,
)

THRESHOLD = timedelta(minutes=60)

@flow
async def alert_on_long_waits_v2(
    slack_block_name: str = "my-slack-webhook",
    task_tags_to_watch: list[str] | None = None,
    check_task_runs: bool = True,
    check_flow_runs: bool = False,  # set True if you want to also flag long-scheduled flow runs (work pool/queue pressure)
):
    slack = await SlackWebhook.load(slack_block_name)
    now = datetime.now(timezone.utc)

    async with get_client() as client:
        if check_task_runs:
            tag_filter = None
            if task_tags_to_watch:
                tag_filter = TaskRunFilterTags(all_=task_tags_to_watch)

            trs = await client.read_task_runs(
                task_run_filter=TaskRunFilter(
                    state=TaskRunFilterState(
                        name=TaskRunFilterStateName(any_=["Pending"])
                    ),
                    tags=tag_filter,
                )
            )
            for tr in trs:
                if tr.state and tr.state.timestamp and (now - tr.state.timestamp) > THRESHOLD:
                    msg_extra = f" | msg: {tr.state.message}" if (tr.state and tr.state.message) else ""
                    await slack.notify(
                        body=(
                            f"Task run '{tr.name}' has been Pending for "
                            f"{(now - tr.state.timestamp)} (likely waiting on tag concurrency). "
                            f"Task run ID: {tr.id} | Flow run ID: {tr.flow_run_id}{msg_extra}"
                        )
                    )

        if check_flow_runs:
            frs = await client.read_flow_runs(
                flow_run_filter=FlowRunFilter(
                    state=FlowRunFilterState(
                        name=FlowRunFilterStateName(any_=["Scheduled"])
                    )
                )
            )
            for fr in frs:
                if fr.state and fr.state.timestamp and (now - fr.state.timestamp) > THRESHOLD:
                    await slack.notify(
                        body=(
                            f"Flow run '{fr.name}' has been Scheduled for "
                            f"{(now - fr.state.timestamp)} (likely due to work pool/queue concurrency). "
                            f"Flow run ID: {fr.id}"
                        )
                    )

Option B — Prefect Cloud 2.x Automations If you’re on Cloud 2.x, you can build an Automation that:

- Watches task-run state changes, - Uses a proactive trigger starting when a task enters Pending, - Expects a state change (e.g., to Running) within 3600 seconds, - Sends a Slack notification if not. Caveat: because 2.x doesn’t expose a unique “awaiting concurrency” state, Pending can mean other things too. To reduce false positives, scope the automation to tasks with the specific concurrency-limited tag(s). Helpful docs (Prefect 2.x): - States in Prefect 2 - Tag-based concurrency limits (tasks) - Automations in Prefect 2 If you can upgrade to Prefect 3.x, you’ll get a cleaner signal: runs blocked by concurrency sit in the AwaitingConcurrencySlot state, which makes Automations and queries much more precise. A couple of questions to tailor this: - Are you on Prefect Cloud or OSS for 2.20.16? - Do you want to monitor task-level tag concurrency only, or also flow runs that are queued due to work pool/queue limits? - Which tag(s) should we scope to? I can adjust the code or automation pattern accordingly.

Thanks for the details — that helps. You’re correct: - In Prefect Cloud 2.20.16, deployment-level (flow run) concurrency is not available — tag-based concurrency is at the task-run level. - When a task is blocked by a tag concurrency limit in 2.x, it stays in Pending (there’s no dedicated AwaitingConcurrencySlot state in 2.x). So the detection pattern is “task run has been Pending for > 60 minutes,” ideally scoped to your concurrency tags to avoid noise. Two solid ways to implement alerts in Cloud 2.20.16: Option A — Cloud Automation (proactive trigger) Use a proactive, event-based Automation so it fires when a task stays Pending too long. UI steps (high-level and safe for 2.x): 1) Create an Automation (Event-based). 2) Scope it to Task runs and filter to your tag pattern. If the UI filter can’t express your pattern cleanly, consider adding a secondary static tag like

concurrency-monitored

to those tasks so you can filter reliably. 3) Make it Proactive: - Start “after” the task run enters Pending. - “Expect” a state change (e.g., to Running or any non-Pending state) within 60 minutes. - Evaluate “for each” task run (so one stuck task doesn’t suppress others). 4) Action: SendNotification using your Slack webhook block. This will send Slack when a task run remains Pending for > 60 minutes. Because Pending can mean a few things in 2.x (not just concurrency), the tag filter is important to keep the alert specific to the tasks governed by your tag-based concurrency. Helpful Prefect 2.x docs: - Automations (2.x) - Task tags and concurrency (2.x) - Events (2.x) Option B — A small monitoring flow you deploy and schedule If you prefer code (or want more control), schedule this flow to run every few minutes. It finds task runs that have been Pending longer than 60 minutes and sends a Slack message. You can scope it to tags matching your pattern (or again, add a stable secondary tag to filter on).

from datetime import datetime, timedelta, timezone

from prefect import flow
from prefect.blocks.notifications import SlackWebhook
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    TaskRunFilter,
    TaskRunFilterState,
    TaskRunFilterStateName,
    TaskRunFilterTags,
)

THRESHOLD = timedelta(minutes=60)

@flow
async def alert_on_long_pending_tasks(
    slack_block_name: str = "my-slack-webhook",
    # e.g. provide a stable tag these tasks carry in addition to your unique tag pattern
    tags_to_watch: list[str] | None = None,
):
    slack = await SlackWebhook.load(slack_block_name)
    now = datetime.now(timezone.utc)

    async with get_client() as client:
        tag_filter = TaskRunFilterTags(all_=tags_to_watch) if tags_to_watch else None

        task_runs = await client.read_task_runs(
            task_run_filter=TaskRunFilter(
                state=TaskRunFilterState(
                    name=TaskRunFilterStateName(any_=["Pending"])
                ),
                tags=tag_filter,
            )
        )

        for tr in task_runs:
            if tr.state and tr.state.timestamp and (now - tr.state.timestamp) > THRESHOLD:
                # Optional: try to reduce false positives by checking message text
                msg_hint = f" | msg: {tr.state.message}" if tr.state.message else ""
                await slack.notify(
                    body=(
                        f"Task run '{tr.name}' has been Pending for "
                        f"{(now - tr.state.timestamp)} (likely waiting on tag concurrency). "
                        f"Task run ID: {tr.id} | Flow run ID: {tr.flow_run_id}{msg_hint}"
                    )
                )

Notes and tips:

- Because Pending can mean “waiting on upstream,” “waiting on infrastructure,” or “waiting on concurrency,” filtering by your concurrency-specific tags is the best way to keep alerts on-target. - If your unique tag pattern isn’t easy to express in the UI filter, add a stable extra tag (e.g.,

concurrency-monitored

) to those tasks so you can filter on it in either Automations or the code above. - You can tune how often tasks retry acquiring a concurrency slot with

PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS

, but that won’t change the 60-minute alert logic. If you eventually upgrade to Prefect 3.x, concurrency waits show up explicitly as the AwaitingConcurrencySlot state, which makes Automations and queries more precise. If you want, share: - The tag prefix/pattern you’re using, - Whether you prefer a UI Automation or a code-based monitor, and I’ll give you a ready-to-apply configuration or a deployment-ready flow tailored to your tags.