Great question. In Prefect 2.20.16, the

wait_for

parameter on

task.submit()

is used to declare explicit upstream dependencies when you want to control execution order but don’t need to pass data between tasks. What it does - Creates an ordering barrier: the submitted task will not start until all items in

wait_for

have finished. - Does not pass data: unlike passing a future as an argument,

wait_for

only controls timing/dependencies, not inputs. What it accepts - A single

PrefectFuture

or a list of `PrefectFuture`s (typically the return values of other

task.submit()

or subflow calls). Failure behavior - If any upstream in

wait_for

finishes in a non-successful terminal state (e.g., failed, cancelled, crashed), the dependent task will not run and will be marked as not executed due to upstream failure. In other words,

wait_for

enforces that upstreams must complete successfully before the downstream will start. Also works with mapping and flows -

wait_for

can be used on

.map()

and on subflow calls. For mapping, the dependency is applied to every mapped child. Examples

python
from prefect import task, flow

@task
def a():
    ...

@task
def b():
    ...

@task
def c():
    # Does not take outputs of a/b; we just want to run after both finish
    ...

@flow
def main():
    fa = a.submit()
    fb = b.submit()
    fc = c.submit(wait_for=[fa, fb])  # c won’t start until a and b complete
    return fc.result()

With mapping:

python
@task
def prepare():
    ...

@task
def process(x):
    ...

@flow
def mapped_flow():
    ready = prepare.submit()
    results = process.map(range(10), wait_for=[ready])  # each mapped child waits for prepare
    return [r.result() for r in results]

Docs - Specifying upstream dependencies with `wait_for`: Specifying upstream dependencies - Tasks and execution: Tasks concept If you have a specific DAG you’re modeling, I’m happy to tailor an example.

In Prefect 2.20.16, task runs can move through these states (state name → state type). Terminal states are marked with a star. - Scheduled → SCHEDULED - Late → SCHEDULED - AwaitingRetry → SCHEDULED - Pending → PENDING - AwaitingConcurrencySlot → AWAITING_CONCURRENCY_SLOT - Running → RUNNING - Retrying → RUNNING - Cancelling → CANCELLING - Cancelled → CANCELLED ★ - Completed → COMPLETED ★ - Failed → FAILED ★ - Crashed → CRASHED ★ Notes - Paused and Suspended are flow-run concepts in 2.x; tasks do not normally enter these states. - “Retrying” indicates an active retry attempt (still RUNNING type); “AwaitingRetry” means the next retry is scheduled but not yet started (SCHEDULED type). - “Late” means the scheduled start time passed and the run did not begin promptly. - “AwaitingConcurrencySlot” appears when task or global concurrency limits are blocking execution. Docs - States overview: Prefect 2: States - Concurrency/limits (for AwaitingConcurrencySlot): Global concurrency limits

Short answer: Not directly. In Prefect 2.20.16 there isn’t a built‑in “TTL” that will automatically fail/cancel a run that’s waiting in AwaitingConcurrencySlot; it will keep retrying at a fixed interval until a slot is available. What you can do - Adjust the poll interval (not a max wait): set

PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS

to change how often Prefect rechecks for a slot (default 30s). - Enforce a hard cap with a flow timeout:

python
  from prefect import flow

  @flow(timeout_seconds=900)  # cancel/fail the flow after 15 minutes total
  def my_flow():
      ...

This will cause the flow to time out if tasks are still waiting for slots. - If you use global concurrency with the concurrency context manager, add an acquisition timeout so tasks don’t wait indefinitely:

python
  from prefect import task
  from prefect.concurrency.asyncio import concurrency

  @task
  async def work():
      async with concurrency("my-limit", timeout_seconds=300):  # give up after 5 minutes
          # do work
          ...

When the timeout elapses, an error is raised and you can handle it (e.g., fail or cancel the flow). - Prefect Cloud Automations: create an automation to cancel/mark failed if a flow run remains Running (or in a given state) longer than X minutes. See States and Global concurrency limits. If you’d like, tell me which concurrency mechanism you’re using (task tag limits vs. the

concurrency()

context vs. work pool concurrency), and your desired timeout, and I can suggest the most precise setup.

Thanks for confirming. For Prefect 2.20.16 task tag concurrency limits: - There is no built-in “max wait/TTL” for the AwaitingConcurrencySlot state. A task that can’t acquire a tag slot will keep retrying indefinitely at a fixed interval. - You can change only the poll interval (not a max wait) via

PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS

(default 30s). Ways to enforce an overall cap anyway - Flow-level timeout (caps total runtime of the flow, including time spent waiting for slots):

from prefect import flow

  @flow(timeout_seconds=900)  # e.g., 15 minutes
  def my_flow():
      ...

- Timeout on the future wait (caps time waiting for a specific task to finish/leave AwaitingConcurrencySlot):

f = my_task.submit()
  try:
      f.result(timeout=900)  # fail after 15 minutes waiting
  except TimeoutError:
      raise RuntimeError("Timed out waiting for concurrency slot")

Note:

@task(timeout_seconds=...)

does not help here because that timeout starts only after the task begins executing; it won’t limit time spent in AwaitingConcurrencySlot. - Prefect Cloud Automations: configure an automation to cancel/fail the flow if it’s Running longer than X minutes. - Consider Prefect’s global concurrency limits for more control (available in 2.x). Using the

concurrency()

context manager lets you set an acquisition timeout:

from prefect import task
  from prefect.concurrency.asyncio import concurrency

  @task
  async def work():
      async with concurrency("my-limit", timeout_seconds=300):  # give up after 5 minutes
          ...

Docs: Global concurrency limits If you want, share how you’re submitting/waiting on your tasks (e.g., calling

.result()

immediately vs. collecting futures) and the timeout you’d like, and I can suggest the cleanest pattern for your DAG.

Short answer: use the future’s state to check the task run state, not the return value. In Prefect 3.x: -

task.submit(...)

returns a

PrefectFuture

. - The task run’s state is available on the future:

task_result.wait(); state = task_result.state

. Then inspect with

state.is_completed()

,

state.is_failed()

, and read

state.message

,

state.name

, etc. -

task_result.result()

returns the function’s return value. If your task literally does

return Completed(...)

, then

task_result.result()

will be that State object as a normal return value; it does not change the run state semantics. So the recommended pattern is:

fut = task.submit(...)
fut.wait()
state = fut.state
if state.is_completed():
    ...

If you really want the terminal state directly (instead of a future), you can:

state = task.submit(..., return_state=True)
if state.is_completed():
    ...

About returning a State from inside a task: - Returning a

Completed(...)

(or any non-failed state) just makes that State the task’s return value; the actual task run will be marked completed if no exception was raised. - If you want to fail the task, raise an exception (preferred) or return a

Failed(...)

state; failure is honored. - There is no “Warning” state type. If you want to surface a warning, consider logging a warning and/or attaching an artifact, or put the warning text in the state’s message if you’re intentionally returning a State as data. Docs: - States in Prefect 3.x - Task runners and futures If your goal is to display a “warning” in the UI for a successful task, can you share what you’d like that to look like? I can suggest the cleanest pattern (e.g., state hook to set a message, logging, or an artifact).

Short answer: yes, the pattern is effectively the same in Prefect 2.20.16. -

task.submit(...)

returns a PrefectFuture - Use

future.result()

to get the task’s return value (data) - Use

future.wait()

to get the task run’s State (you can also read

future.state

after it finishes) Returning a State from inside a task: - If your task does

return Completed(...)

(or any other non-failed State), that State object is just the task’s return value. It does not become the task run’s state. The task run itself will be Completed if no exception was raised. - To fail a task, raise an exception; that’s the recommended way in 2.x. There isn’t a “Warning” state type in 2.x either—log a warning or attach an artifact if you want to surface a warning. Example (2.20.16):

fut = my_task.submit(...)
state = fut.wait()          # State object for the task run
if state.is_completed():
    value = fut.result()    # the task's return value
else:
    # handle failure/other states

Docs: - Prefect 2.x: States - Prefect 2.x: Task runners and futures If you can upgrade, Prefect 3.x keeps the same mental model and adds some ergonomics (e.g., a

return_state

option on

submit

). Happy to share a 2.x-to-3.x migration snippet if helpful.

Mostly yes, but with an important caveat. Your snippet will run in Prefect 2.20.16:

fut = my_task.submit(...)
state = fut.wait()          # State object for the task run
if state.is_completed() and "Warning" not in str(state.name or ""):
    value = fut.result()    # the task's return value
else:
    # handle failure/other states

However, checking for "Warning" in the task run state’s name likely won’t do what you intend: - There is no Warning state type in Prefect 2.x; the run state name for a successful task is typically "Completed" (or sometimes "Cached"). It won’t become "Warning" just because your task returned

State(type=COMPLETED, name="Warning")

— that returned State is just the task’s return value, not the run state. - In other words, if you’re returning a State from inside the task to carry a warning, you must inspect the task’s return value, not the run state. If your task returns a State object to signal a soft warning, do this instead:

fut = my_task.submit(...)
state = fut.wait()
if not state.is_completed():
    # handle failure/other states
    ...

out = fut.result()
if isinstance(out, State) and (out.name == "Warning" or "Warning" in (out.message or "")):
    # handle your warning case
else:
    # proceed with normal result

If your goal is to surface a warning in the UI for a successful task, a simpler and more idiomatic pattern is to log a warning from the task and return the real data:

from prefect import task, get_run_logger

@task
def my_task(...):
    logger = get_run_logger()
    ...
    logger.warning("Something noteworthy happened")
    return real_result

References: - https://docs-2.prefect.io/latest/concepts/states/ - https://docs-2.prefect.io/latest/concepts/task-runners/

Great question. In Prefect 2.20.16 you have a few good options, depending on whether you want the task to truly fail (for observability) or just “soft-fail”. Option A — Catch inside the task and log a warning (flow and task stay Completed)

from prefect import task, get_run_logger

@task
def my_task(...):
    logger = get_run_logger()
    try:
        # do work
        return result
    except Exception as e:
        logger.warning(f"My task soft-failed: {e}")
        return None  # or a sentinel/default value

- Easiest way to keep the flow green and continue. - You’ll get a warning in logs, but the task itself won’t be marked Failed. Option B — Let the task fail, but don’t block downstream or the flow 1) Allow downstream tasks to run even if the upstream failed using `allow_failure`:

from prefect import flow, task, get_run_logger, allow_failure, State

@task
def may_fail():
    ...

@task
def consume(upstream):
    logger = get_run_logger()
    # When using allow_failure, `upstream` will be a State if the task failed
    if isinstance(upstream, State) and upstream.is_failed():
        logger.warning("Upstream failed; continuing with defaults")
        return "default"
    return upstream  # if upstream completed, this is the actual value

@flow
def my_flow():
    fut = may_fail.submit()
    safe = allow_failure(fut)       # downstream will run even if `fut` failed
    processed = consume.submit(safe)
    return processed

2) If you need to inspect the failure in the flow without raising, use:

exc_or_value = fut.result(raise_on_failure=False)
# If the task failed, this returns the exception object; otherwise it returns the value

Notes - If you pass a failed future directly as a parameter to another task (without

allow_failure

), Prefect 2 will skip the downstream task due to the failed dependency. - To keep the flow green, avoid returning failed futures/states from the flow itself; handle them and return a normal value/state. - If your goal is visibility in the UI without failing the flow, Option B is nice: the task run shows Failed (good observability), but your flow still proceeds. References - States (Prefect 2.x) - Task runners and futures (Prefect 2.x) - Specifying upstream dependencies (covers patterns like allowing failed upstreams) If you share a small snippet of your current flow, I can tailor the pattern to your exact wiring.

It will work, but it’s a bit brittle. What your snippet relies on is treating a Prefect

State

as the task’s return value (data). Prefect won’t apply that returned state to the task run; it’ll still mark the task run Completed if no exception was raised. So your check:

fut = my_task.submit(...)
out = fut.result()
if isinstance(out, State) and (out.name == "Warning" or "Warning" in (out.message or "")):
    # handle your warning case
else:
    # proceed with normal result

will function as long as your task consistently returns a

State

when there’s a warning. However, consider the trade-offs: - Type inconsistency: sometimes your result is a

State

, other times it’s your real value, which can make downstream code messy. - Observability: the “warning” lives only in the returned value; the task run state/message in the UI won’t show it. - Using

State

as data is unconventional in 2.x and can surprise future readers. Cleaner patterns in Prefect 2.20.16: 1) Return structured data with a warning field

@task
def my_task(...):
    ...
    if issue:
        return {"value": value, "warning": "Something noteworthy"}
    return {"value": value, "warning": None}

out = my_task.submit(...).result()
if out["warning"]:
    # handle warning
else:
    # proceed

2) Log a warning (and optionally create an artifact) and return the normal value

from prefect import task, get_run_logger

@task
def my_task(...):
    logger = get_run_logger()
    ...
    if issue:
        logger.warning("Something noteworthy happened")
    return value

This keeps types consistent and surfaces the warning in logs/UI. 3) If you truly want to treat the condition as a failure but keep the flow running, don’t raise in the task; instead, in the flow handle the future without raising:

fut = my_task.submit(...)
val_or_exc = fut.result(raise_on_failure=False)
if isinstance(val_or_exc, Exception):
    # treat as warning and continue
else:
    # proceed with val_or_exc

Docs for reference: - States (Prefect 2.x) - Task runners and futures (Prefect 2.x) If you still prefer the “return State to signal warning” approach, I’d recommend making it consistent (always return a State, even on success) to avoid type juggling. But in general, a structured result or logging is the more maintainable route.

Both return a bool. -

state.is_failed()

-> bool -

state.is_completed()

-> bool Example:

if state.is_failed():
    ...

Short answer:

.result()

gives you the task’s return value (or raises the task’s exception by default), not the State. Details: - With `return_state=True`: -

task.submit(..., return_state=True)

returns a

State

object directly (so your variable isn’t a future). - Calling

state.result()

returns the task’s return value. On failure it raises by default; use

state.result(raise_on_failure=False)

to get the exception object instead. - You can inspect status via

state.is_completed()

/

state.is_failed()

. - With the default (

return_state=False

): -

task.submit(...)

returns a

PrefectFuture

. -

future.result()

returns the task’s return value (raises on failure by default). - Inspect the state via

future.state.is_completed()

/

future.state.is_failed()

. Examples:

# return_state=True -> you get a State
state = task.submit(x=1, return_state=True)
assert state.is_completed()
value = state.result()  # the task’s return value

# default -> you get a Future
future = task.submit(x=1)
value = future.result()  # the task’s return value
if future.state.is_failed():
    ...

Refs: - States in Prefect 3 - Task runners and futures

Short answer: It can be COMPLETED, but you should not rely on it. -

state = task.submit(..., return_state=True)

returns a State snapshot at submit time. Typically that’s a non-terminal state (e.g., PENDING/SUBMITTED/RUNNING). If the task finishes very quickly, you might see COMPLETED, but it’s not guaranteed. - If you want a reliable terminal state, don’t use

return_state=True

with submit. Instead: - Get a future and wait:

fut = task.submit(x=1)
    value = fut.result()  # blocks until done
    assert fut.state.is_completed()

- Or wait explicitly:

from prefect.futures import wait
    fut = task.submit(x=1)
    wait([fut])
    assert fut.state.type.name == "COMPLETED"

- If you specifically want a State (not a Future) and you want it to be terminal, run synchronously:

state = task(x=1, return_state=True)  # runs now; returns terminal State
  assert state.is_completed()

Refs: - States in Prefect 3 - Task runners and futures

Yes. In Prefect 3.x you can alert on this using an Automation that triggers when a flow run enters a Failed (or Crashed) state, and optionally filter on the state message so you only alert on this specific retrieval failure. This works the same for Kubernetes workers. Why this works - When the worker tries to load your deployment’s flow (e.g., from Git/S3/filesystem) and cannot, the flow run transitions to a terminal state (typically Failed) with a state message like “Flow could not be retrieved from deployment.” - Automations can watch for those state changes and send a notification. How to set it up (UI) 1) Create a Notification block (Slack/email/webhook) - In the UI, go to Blocks → Add Block → choose your channel (e.g., Slack Webhook), configure, and save. - Docs: Notification blocks 2) Create an Automation - Go to Automations → New Automation. - Trigger: Flow run state entered. - States to watch: select Failed (and optionally Crashed). - Scope (optional): narrow to specific deployment(s) and/or your Kubernetes work pool. - Advanced filter (optional but recommended): State message contains: Flow could not be retrieved from deployment - Action: Send a notification → select the Notification block you created. - Docs: Automations Notes - If your UI does not show an “Advanced” or “message contains” filter, just alert on Failed/Crushed for the relevant deployment or work pool. You’ll still catch these retrieval failures (they’ll surface as Failed with that message). - This failure commonly indicates issues like an incorrect entrypoint, missing ref/branch, auth/permissions to your source, or network access from the worker to your storage. - Kubernetes-specific setup doesn’t change the Automation; the worker’s failure to load the flow still results in a flow-run state change that the Automation can catch. - More background on events/state transitions: Events in Prefect Want me to draft the Automation criteria for a specific deployment or set up a Slack notification block with you? Also, are you on Prefect Cloud or OSS? I can tailor the steps.

Yes—on 2.20.16 you can alert on these using an Automation that triggers when a flow run enters Failed (and/or Crashed). One caveat: in Prefect 2.x you cannot filter the Automation trigger by the state message text, so you can’t pre-filter specifically on “Flow could not be retrieved from deployment.” You can still include the state message in the notification so it’s clear why it failed. Why this works - When the Kubernetes worker can’t load your flow from the deployment’s source, the flow run is created and quickly moves to Failed with a message like “Flow could not be retrieved from deployment.” - Automations can watch for those state changes and notify you. Flow-level hooks won’t help here because the flow never loads. How to set it up (UI in Prefect 2.x) 1) Create a Notification block (Slack/email/webhook) - Blocks → Add Block → choose your channel (e.g., Slack Webhook), configure, save. - Docs: Notification blocks 2) Create an Automation - Automations → New Automation → “Flow run state change”. - When state enters: select Failed (and optionally Crashed). - Scope: narrow by Deployment and/or your Kubernetes Work Pool (to avoid noise). - Action: “Send a notification” → pick the Notification block. - Message template (include the state message so you can see the retrieval error in the alert):

Flow run {{ flow_run.name }} ({{ flow_run.id }}) entered {{ flow_run.state.type }} / {{ flow_run.state.name }}
  Message: {{ flow_run.state.message }}
  Deployment: {{ deployment.name }} | Work pool: {{ work_pool.name }}

Notes - Prefect 2.20.16 does not support filtering Automations by the state message text; use scoping (deployment/work pool/tags) to minimize false positives and rely on the included message for context. - This works the same whether you’re on Cloud or OSS. - Common causes: wrong entrypoint, missing/incorrect ref/branch, missing credentials for your source, or worker network access to storage. If you tell me your deployment/work pool names (and whether you’re on Cloud or OSS), I can outline the exact Automation settings to use.

Got it—thanks for the context. In Prefect Cloud (2.20.16), the fact that your workers run on your own infrastructure doesn’t change how Automations and notifications work: Automations run in Cloud and select a specific Notification block, which determines the Slack destination. So the Slack channel is effectively static per Automation. Practical options: - One block per workspace (recommended) - In each Cloud workspace, create a Slack Webhook block pointing to that workspace’s channel (e.g., #prefect-alerts-dev, #prefect-alerts-prod). - Create the same Automation in each workspace and select that workspace’s block. No code changes; nice separation by workspace. - Route by environment/tags within a single workspace - Create multiple Automations scoped by deployment/work-pool/tags (e.g., tag env:prod vs env:dev), each pointing to a different Slack Webhook block/channel. - Dynamic routing via your own webhook (advanced) - Use a single Automation action to call your own webhook/service. Your service inspects the payload (deployment name, tags, work pool, etc.) and forwards to the correct Slack channel using Slack’s API. This avoids storing multiple Slack webhooks in Prefect Cloud and gives you full routing control. Notes: - Since Automations run in Cloud, the Slack message is sent from Prefect Cloud to Slack; worker location doesn’t impact delivery. - If you’d rather not store Slack webhooks in Prefect Cloud, the webhook-proxy pattern is the best fit. - Flow retrieval failures still won’t be capturable by in-flow hooks (the flow never loads); Automations are the right tool. If you share how many workspaces/environments you have and which channels you want, I can outline the exact set of blocks/automations to create and a message template to include the state message.

Absolutely—here’s how the webhook-proxy pattern works in Prefect 2.20.16. Goal - Use one Automation in Prefect Cloud to send a single HTTP POST (via a Webhook Notification block) to your own endpoint. - Your endpoint inspects the payload (e.g., deployment name, tags, work pool, or a custom header) and forwards the alert to the correct Slack channel. This gives you dynamic routing outside of Prefect. High-level flow 1) Prefect Automation (Cloud) triggers on flow-run Failed/Crash. 2) Action: Send a notification → Webhook block → POSTs to your proxy URL. 3) Your proxy decides the Slack channel (based on environment, workspace, tags, etc.) and calls Slack chat.postMessage (or forwards to the right Slack incoming webhook). Why this pattern - Prefect 2.x Automations choose a single Notification block per action (static). - The generic Webhook Notification block lets you send to any HTTP endpoint you control, and you can add static headers and a templated body. What you’ll create A) A Webhook Notification block in each workspace - URL: your proxy endpoint (e.g., https://alerts.example.com/prefect/notify) - Method: POST - Headers: include a shared secret (e.g., Authorization: Bearer <secret>) and, optionally, a workspace label you set per workspace, like X-Workspace: prod or X-Workspace: dev. This gives your proxy an easy, reliable “workspace” signal without relying on template variables. - Body template: include the core details you’ll want for routing and the Slack message. Example Automation body template Use Jinja to render a JSON payload your proxy can parse. If you run into escaping issues with complex messages, you can simplify or switch to a text payload.

{
  "event": "flow_run_state_change",
  "state_type": "{{ flow_run.state.type }}",
  "state_name": "{{ flow_run.state.name }}",
  "state_message": "{{ flow_run.state.message | default('') }}",
  "flow_run_id": "{{ flow_run.id }}",
  "flow_run_name": "{{ flow_run.name }}",
  "deployment_name": "{{ deployment.name if deployment else '' }}",
  "work_pool_name": "{{ work_pool.name if work_pool else '' }}",
  "tags": "{{ flow_run.tags | join(',') }}"
}

B) An Automation in each workspace - Trigger: Flow run state entered → Failed (and/or Crashed). - Scope: Narrow by deployment/work pool/tags as needed. - Action: Send a notification → Select your Webhook Notification block. C) Your proxy service - Receives the POST, validates the shared secret, decides the Slack channel, and posts the message to Slack. Minimal FastAPI proxy example ``` import os from fastapi import FastAPI, Request, HTTPException from slack_sdk import WebClient from slack_sdk.errors import SlackApiError app = FastAPI() SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"] SHARED_SECRET = os.environ["PREFECT_WEBHOOK_SECRET"] client = WebClient(token=SLACK_BOT_TOKEN) # Simple example mapping based on header or payload fields CHANNEL_MAP = { "dev": "#prefect-alerts-dev", "staging": "#prefect-alerts-staging", "prod": "#prefect-alerts-prod", } def pick_channel(headers, payload): # Prefer a header set per workspace block (reliable) ws = headers.get("x-workspace", "").lower() if ws in CHANNEL_MAP: return CHANNEL_MAP[ws] # Fallbacks based on payload content tags = set([t.strip().lower() for t in payload.get("tags", "").split(",") if t]) if "env:prod" in tags: return CHANNEL_MAP["prod"] if "env:staging" in tags: return CHANNEL_MAP["staging"] return CHANNEL_MAP.get("dev", "#general") @app.post("/prefect/notify") async def prefect_notify(request: Request): # Verify shared secret auth = request.headers.get("authorization", "") if not auth.startswith("Bearer ") or auth.split(" ", 1)[1] != SHARED_SECRET: raise HTTPException(status_code=401, detail="Unauthorized") payload = await request.json() channel = pick_channel(request.headers, payload) # Construct Slack message text = (

f"Prefect flow run {payload.get('flow_run_name')} " f"({payload.get('flow_run_id')}) entered " f"{payload.get('state_type')}/{payload.get('state_name')}.\n" f"Deployment: {payload.get('deployment_name')} | Work pool: {payload.get('work_pool_name')}\n" f"Message: {payload.get('state_message','')}" ) try: client.chat_postMessage(channel=channel, text=text) except SlackApiError as e: # Log and re-raise or swallow based on your preference raise HTTPException(status_code=500, detail=f"Slack error: {e.response['error']}") return {"ok": True} ``` Security and deployment tips - Authentication: Use a shared secret header (as above). You can add an additional IP allowlist or signature scheme if you want. - TLS: Terminate TLS at your proxy or a fronting load balancer (e.g., API Gateway, NGINX, Cloudflare). - Hosting: Easiest paths are a lightweight container in your infra (Kubernetes/VM) with a public URL, or a serverless function (AWS API Gateway + Lambda). - Secrets: Keep your Slack bot token out of Prefect; store it where your proxy runs (env var, secret manager). - Reliability: Add basic retries to Slack posting; consider dead-lettering to a queue if Slack fails. How this solves “dynamic channel per workspace” - Each Prefect Cloud workspace points to the same proxy URL but sets a different header (e.g., X-Workspace: prod/dev). - Your proxy uses that header (or tags) to select the channel, so routing is dynamic without changing Automations. Docs for reference - Automations (2.x): Automations - Notification blocks (including Webhook): Notification blocks If you share your environment naming (dev/staging/prod) and preferred channels, I can draft the exact header values and a ready-to-paste Webhook block payload template for your Automation.

Short answer: By default, yes—if task_1 fails and task_2 has

wait_for=[task_1]

, task_2 will not run. It will be cancelled due to a failed upstream dependency. How to run task_2 anyway - Use

allow_failure

to mark the upstream dependency as “non-blocking”:

from prefect import flow, task, allow_failure

  @task
  def task_1():
      raise ValueError("boom")

  @task
  def task_2():
      return "I still ran"

  @flow
  def main():
      a = task_1.submit()
      # Option A: keep explicit dependency but allow failure
      b = task_2.submit(wait_for=[allow_failure(a)])
      # Option B: pass as input; allow_failure also prevents blocking here
      # b = task_2.submit(upstream=allow_failure(a))
      return b

Notes - Without

allow_failure

, any failed upstream in

wait_for

will cancel the downstream. - If the upstream is retrying, the downstream will wait; if the upstream ultimately fails, the downstream is cancelled; if it succeeds, the downstream runs. - Another pattern is to catch exceptions inside

task_1

and return a sentinel value instead of failing; then the task completes and downstream runs, but you must handle the sentinel in

task_2

. Docs - Upstream dependencies: Specifying upstream dependencies - States: States in Prefect If you describe your intended behavior (run regardless vs. run only on specific failures), I can suggest the cleanest pattern.

Short answer: yes—what you’re seeing is expected. What happens in Prefect 3.x with `wait_for`: - A task run is always created first. That’s why you see “Created task run 'Send Event-0' for task 'Send Event'”. - Immediately after creation, the engine resolves parameters and checks dependencies (

wait_for

). If any of the upstreams did not reach a COMPLETED state (and you didn’t mark them with

allow_failure

), the downstream task is put into a Pending state named “NotReady” and the task function is not invoked. - The “Executing 'Send Event-0' immediately…” log line is emitted as the engine begins orchestration for that run in-process. The dependency check happens right after; so you can see this message even if the task never actually executes its body because

wait_for

fails the gate. So your mental model is right: the downstream run is created, then it checks

wait_for

, then decides not to run and ends in Pending/NotReady if an upstream failed. Tips: - Passing a future as a parameter implicitly adds it to

wait_for

, so you don’t need both unless you’re gating on a future you don’t pass as an argument. - If you do want the downstream to proceed even when an upstream fails, wrap that dependency in `allow_failure`:

from prefect import task, flow, allow_failure

@task
def up():
    raise RuntimeError("boom")

@task
def down():
    return "ran anyway"

@flow
def f():
    u = up.submit()
    # This will run even if `u` failed
    d = down.submit(wait_for=[allow_failure(u)])

If you can share your Prefect version and task runner (e.g., Concurrent/ThreadPool, ProcessPool, a remote worker), I can pinpoint exactly where that “Executing … immediately” line is coming from in your setup.