Short answer: your trigger is too loose on the primary resource and may be getting suppressed by loop-safety or just not matching what you expect. A few concrete issues and fixes: What’s likely wrong - match is empty. While allowed, it’s safer to scope the primary resource to flow-run events: add

prefect.resource.id: prefect.flow-run.*

. Without it, you rely entirely on

expect

for scoping and it can behave unexpectedly. - for_each should be a list, not a set. Use

["prefect.resource.id"]

, not

{"prefect.resource.id"}

. - Loop-safety suppression: If the deployment you’re running belongs to the same flow you’re matching (the same

prefect.flow.<id>

), Prefect will suppress the action to prevent an infinite loop. Make sure the target deployment is for a different flow. - Targeting by flow vs deployment: You’re matching the related flow id. If that flow has multiple deployments, you’ll trigger off any of them. It’s more robust to match the related deployment you care about. - Sanity check the flow id: Ensure

prefect.flow.5b93455b-d019-48c0-8463-4748b9c8672a

is the Flow id (not a Flow Run id). The primary event is

prefect.flow-run.Completed

, and the related resource with role

flow

must match that flow id. A corrected, safer trigger - Scopes to flow-run events - Buckets per flow run - Filters specifically by the upstream deployment (recommended)

Automation(
    name="Run Zoho Import after Scheduling Flow Completion",
    description="Trigger Zoho import flow after the scheduling deployment completes",
    tags=["v25.02.2026"],
    trigger=EventTrigger(
        # Scope primary resource to flow-run events
        match={"prefect.resource.id": "prefect.flow-run.*"},
        # Prefer filtering by the upstream deployment; replace with your upstream deployment id
        match_related={
            "prefect.resource.role": "deployment",
            "prefect.resource.id": "prefect.deployment.<UPSTREAM_DEPLOYMENT_ID>"
        },
        expect={"prefect.flow-run.Completed"},
        # Must be a list, not a set
        for_each=["prefect.resource.id"],
    ),
    actions=[
        RunDeployment(
            # Ensure this deployment is for a different flow than the one you filter on above
            deployment_id=UUID("<DOWNSTREAM_DEPLOYMENT_ID>"),
            parameters={
                "bucket": "google-maps-scrape-data-scheduling",
                "update_tags": True,
            },
            schedule_after=timedelta(0),
        )
    ],
)

If you really want to match by flow (any deployment of that flow), use:

match_related={
    "prefect.resource.role": "flow",
    "prefect.resource.id": "prefect.flow.5b93455b-d019-48c0-8463-4748b9c8672a",
}

but keep the

match={"prefect.resource.id": "prefect.flow-run.*"}

and

for_each=["prefect.resource.id"]

. Diagnostic checklist - Is the automation enabled? - In the Automations page, does the automation show any incidents or “suppressed due to loop-safety” notes? - In Events, can you see

prefect.flow-run.Completed

for the upstream run with a related resource

role=deployment

(or

role=flow

) matching your filter? - Is the downstream deployment in a running work pool with a healthy worker? - Are the action parameters accepted by the downstream flow? If you share: - The upstream deployment id (or confirm it’s a different flow than the downstream), - A screenshot or JSON of one upstream

prefect.flow-run.Completed

event (including related resources), - Whether the automation shows as “Triggered” or “Suppressed” in the Automations UI, …I can pinpoint the exact reason it didn’t fire and give you a precise fix.

Great question. In self-hosted Prefect 3.x, “automation logs” are best viewed via events and server logs. Here are the practical ways to inspect them: CLI (quickest) - List/inspect automations (config only, not logs):

prefect automation ls
prefect automation inspect "my-automation"

- Stream automation-related events (these show when an automation was triggered, actions executed/failed, etc.):

# All automation events
prefect events stream --format json | jq 'select(.event | startswith("prefect.automation"))'

# For a specific automation by ID
prefect events stream --format json | jq 'select(.resource.id=="prefect.automation.<AUTOMATION_ID>")'

# Only action failures
prefect events stream --format json | jq 'select(.event=="prefect.automation.action.failed")'

Event names to look for: -

prefect.automation.triggered

/

prefect.automation.resolved

-

prefect.automation.action.triggered

/

prefect.automation.action.executed

/

prefect.automation.action.failed

UI (if you run the OSS UI) - Go to the Automations page to verify the automation exists and is enabled. - Open the Events page and filter: - By event prefix:

prefect.automation

- Or by resource id:

prefect.automation.<AUTOMATION_ID>

Server/service logs (helpful for deeper debugging) - If you run the server locally:

# Increase verbosity
prefect server start --log-level DEBUG

Watch the output for lines from triggers/actions services indicating automations being evaluated and actions being executed/failed. - If you run in Docker/Kubernetes, check container/pod logs for the API and services; increase log level with

PREFECT_SERVER_LOGGING_LEVEL=DEBUG

. REST API (scriptable/history) - Query automation events directly:

curl -X POST http://<your-api-host>/api/events/filter \
  -H "Content-Type: application/json" \
  -d '{
    "filter": {
      "event": { "prefix": ["prefect.automation"] },
      "resource": { "id": { "values": ["prefect.automation.<AUTOMATION_ID>"] } }
    },
    "limit": 200
  }'

Notes - There isn’t a

prefect automation incidents

CLI command; use the events stream/API to see when automations triggered and how actions performed. - If you don’t see any events, confirm: - The automation is enabled. - The trigger conditions actually match incoming events. - Your work pool/worker is healthy (for RunDeployment actions). - Server log level is high enough to show trigger evaluation. If you share how you’re hosting (local

prefect server start

, Docker, or Helm) and your automation ID, I can give you exact commands/filters tailored to your setup.

Yes—automations are fully supported in Prefect 3.x OSS (self‑hosted). You just need to ensure the background services that evaluate triggers and run actions are running. Quick ways to enable/run them - Easiest (single process): start the server with services enabled

prefect server start

This includes the background services for events, triggers (reactive/proactive), scheduling, and actions. - Scaled/HA (separate API and services): - Run API(s) without services

prefect server start --no-services

- Run background services once (separate process/container)

prefect server services start

Verify services

prefect server services ls

Look for triggers/services like: - Reactive/Proactive Triggers - Event Persister - TaskRunRecorder - (Optional) EventLogger Useful env vars (optional) - Control proactive evaluation cadence:

PREFECT_EVENTS_PROACTIVE_GRANULARITY=5   # seconds

- Enable/disable specific services (defaults are generally fine):

PREFECT_API_SERVICES_TRIGGERS_ENABLED=true
PREFECT_API_SERVICES_EVENT_PERSISTER_ENABLED=true

- Increase verbosity for debugging:

PREFECT_SERVER_LOGGING_LEVEL=DEBUG

How to confirm automations are executing - Stream automation events

prefect events stream --format json | jq 'select(.event | startswith("prefect.automation"))'

You’ll see: - prefect.automation.triggered / resolved - prefect.automation.action.triggered / executed / failed - Inspect automation definitions

prefect automation ls
prefect automation inspect "your-automation-name"

If you’re not seeing automations fire: - Make sure at least one instance of background services is running (either via

prefect server start

or

prefect server services start

) - Ensure the automation is enabled (status in

prefect automation ls

) - Check server logs with DEBUG for triggers/actions - Confirm your trigger filters match the events you expect If you share how you’re running Prefect (single node vs Kubernetes/Docker) I can suggest the exact commands/docker-compose/Helm values to enable services cleanly.

Do you mean the Events feed in the UI or the CLI stream? Here are the best ways to make it “more verbose” in both: CLI (most control) - Use JSON format (includes full payload, labels, and related resources). The text format is intentionally terse.

# Full event JSON
prefect events stream --format json

# Pretty-print and filter with jq (examples)
prefect events stream --format json | jq .

# Show only automation events, fully expanded
prefect events stream --format json | jq 'select(.event | startswith("prefect.automation"))'

# Show event, resource, related, and payload fields neatly
prefect events stream --format json | jq '{event, occurred, resource, related, payload}'

Tip: Pipe to a file if you want to analyze later:

prefect events stream --format json --output-file events.jsonl

Server-side verbosity (self-hosted) - Enable the EventLogger service to log every event received by the API:

export PREFECT_API_SERVICES_EVENT_LOGGER_ENABLED=true
# then start the server with services
prefect server start
# or if you separate API/services:
prefect server services start

- Increase server log verbosity (includes trigger evaluation/action logs):

prefect server start --log-level DEBUG
# or
export PREFECT_SERVER_LOGGING_LEVEL=DEBUG
prefect server start

UI - Open the Events page, click an event to view the details panel. That shows the full JSON, related resources, and payload fields (more verbose than the list row). - Filter by event prefix (e.g.,

prefect.automation

) or resource id to focus on what you need. If you tell me which stream you’re using (UI vs CLI) and what details you’re missing (e.g., related resources, payloads, action outcomes), I can give you an exact jq filter or config to surface it.

A few common reasons you won’t see “prefect.flow-run.Running” in the Events feed (and how to fix each): 1) The event never got emitted - A Running event is only emitted after a successful state transition to Running. If the run went straight from Pending to Completed (very fast local run) or the state validation didn’t complete, you may miss a separate Running event. - Emission disabled at source: the worker/client only emits events when it can reach your API. - Ensure your workers/flows have

PREFECT_API_URL

pointing at your self-hosted API (and not running completely offline). - Quick check in a worker env:

python -c "from prefect.events.worker import should_emit_events; print(should_emit_events())"

Expect

True

. 2) Event published but not persisted (OSS service side) - EventPersister service must be running to store events in the DB. It’s enabled by default, but verify: - If running a single process:

prefect server start

- If separating API from services:

prefect server start --no-services
    prefect server services start

- Check for persistence warnings in server logs at DEBUG level: - “Event queue full … dropping event” (increase queue:

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_QUEUE_MAX_SIZE

) - “Max flush retries reached, dropping events” (DB issues) - Event retention: by default, events older than ~7 days are pruned (

PREFECT_SERVER_EVENTS_RETENTION_PERIOD

). Old events won’t appear. 3) It exists, but your UI/stream isn’t showing it - UI filter/time window: Make sure the Events page filter includes

prefect.flow-run.*

and the time window covers the moment the run started. The WebSocket “backfill” only brings ~15 minutes by default. - CLI stream with full detail:

prefect events stream --format json | jq 'select(.event=="prefect.flow-run.Running")'

Or for one specific flow run:

prefect events stream --format json | jq 'select(.resource.id=="prefect.flow-run.<RUN_ID>")'

4) Ephemeral/local runs - If you used an ephemeral API (no persistent DB) and the server stopped, historical events are gone. - Ensure you’re running a persistent server (e.g.,

prefect server start

) and not shutting it down before checking events. 5) Overly large event payload (rare) - Extremely large state messages can push event size over the limit (~1.5 MB) and get dropped. You’d see an error in server logs. This is configurable via

PREFECT_SERVER_EVENTS_MAXIMUM_SIZE_BYTES

. Quick validation checklist - Server/services running:

prefect server services ls

- Worker can emit: run the

should_emit_events()

snippet above - Stream raw events now:

prefect events stream --format json | jq '{event, occurred, resource} | select(.event|startswith("prefect.flow-run"))'

- Check server logs with DEBUG:

prefect server start --log-level DEBUG

If you share how you’re running (local process vs Docker/K8s), your

PREFECT_API_URL

, and one flow run ID/time window, I can help pinpoint exactly where the Running event is being lost.

You got it — here’s a minimal, end‑to‑end example that creates a simple automation in Python that runs a deployment whenever any flow run completes. This uses the high‑level Automation.create() API in Prefect 3.x. Replace the placeholders with your actual IDs and names before running.

from datetime import timedelta
from uuid import UUID

from prefect.events import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import RunDeployment

# 1) Set your downstream deployment ID (the one you want to run)
DOWNSTREAM_DEPLOYMENT_ID = UUID("00000000-0000-0000-0000-000000000000")

# 2) (Optional but recommended) Narrow to a specific upstream deployment by ID
UPSTREAM_DEPLOYMENT_ID = UUID("11111111-1111-1111-1111-111111111111")

# Build the automation
automation = Automation(
    name="Run downstream when upstream completes",
    description="Runs a selected deployment when the upstream deployment's flow run completes",
    enabled=True,
    trigger=EventTrigger(
        # Watch for flow-run completion events
        expect={"prefect.flow-run.Completed"},
        posture="Reactive",
        # Scope primary resource to flow-run events
        match={"prefect.resource.id": "prefect.flow-run.*"},
        # Scope to a specific upstream deployment (recommended to avoid accidental loops)
        match_related={
            "prefect.resource.role": "deployment",
            "prefect.resource.id": f"prefect.deployment.{UPSTREAM_DEPLOYMENT_ID}",
        },
        # Evaluate per flow run
        for_each=["prefect.resource.id"],
    ),
    actions=[
        RunDeployment(
            # Ensure this is a different deployment than the upstream
            source="selected",
            deployment_id=DOWNSTREAM_DEPLOYMENT_ID,
            # Optional: pass parameters and/or delay
            parameters={"example_param": "value"},
            schedule_after=timedelta(seconds=0),
        )
    ],
).create()

print(f"Created automation: {automation.name} (id={automation.id})")

Notes to ensure it works - Make sure your server/services are running (needed for evaluation and actions): - Single process:

prefect server start

- Split:

prefect server start --no-services

and

prefect server services start

- Confirm the automation shows as enabled: -

prefect automation ls

-

prefect automation inspect "Run downstream when upstream completes"

- Validate events are visible: -

prefect events stream --format json | jq 'select(.event | startswith("prefect.flow-run"))'

- Avoid loops: the upstream and downstream deployments should be different flows/deployments. - If you prefer to match by flow instead of deployment, change

match_related

to:

match_related={
      "prefect.resource.role": "flow",
      "prefect.resource.id": "prefect.flow.<FLOW_ID>"
  }

If you share your upstream/downstream deployment IDs and how you’re hosting Prefect (local vs Docker/K8s), I can tailor this snippet exactly to your setup.

Perfect — here’s a minimal, copy‑paste local example with two flows and an Automation that runs the downstream deployment when the upstream completes. This uses Prefect 3.x and a local process work pool. 1) Create flows (flows.py)

from prefect import flow

@flow(log_prints=True, name="upstream")
def upstream_flow():
    print("Upstream is running...")
    return 42

@flow(log_prints=True, name="downstream")
def downstream_flow():
    print("Downstream triggered by automation!")

2) Register deployments and create the automation (setup.py)

from datetime import timedelta
from uuid import UUID

from flows import upstream_flow, downstream_flow
from prefect.events import Automation
from prefect.events.schemas.automations import EventTrigger
from prefect.events.actions import RunDeployment

# 1) Create deployments in the local "process" work pool
UP_DEPLOYMENT_ID = upstream_flow.deploy(
    name="up-deploy",
    work_pool_name="local",   # make sure this pool exists
)

DOWN_DEPLOYMENT_ID = downstream_flow.deploy(
    name="down-deploy",
    work_pool_name="local",
)

print(f"Upstream deployment id: {UP_DEPLOYMENT_ID}")
print(f"Downstream deployment id: {DOWN_DEPLOYMENT_ID}")

# 2) Create an automation: when upstream completes, run downstream
automation = Automation(
    name="run-downstream-on-upstream-complete",
    description="Run downstream when upstream flow-run completes",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Completed"},
        posture="Reactive",
        # Scope to flow-run events
        match={"prefect.resource.id": "prefect.flow-run.*"},
        # Filter to the specific upstream deployment (avoid loops/false matches)
        match_related={
            "prefect.resource.role": "deployment",
            "prefect.resource.id": f"prefect.deployment.{UP_DEPLOYMENT_ID}",
        },
        # Evaluate per originating flow run
        for_each=["prefect.resource.id"],
    ),
    actions=[
        RunDeployment(
            source="selected",
            deployment_id=UUID(str(DOWN_DEPLOYMENT_ID)),
            parameters={},                 # optional
            schedule_after=timedelta(0),   # run immediately
        )
    ],
).create()

print(f"Created automation: {automation.name} (id={automation.id})")

3) Run everything locally - Terminal 1: start the server

prefect server start

- Terminal 2: create a process work pool (skip if you already have one named “local”)

prefect work-pool create local --type process

- Terminal 2: start a worker on that pool

prefect worker start -p local

- Terminal 3: register deployments and create the automation

python setup.py
prefect deployment ls

- Terminal 3: kick off the upstream deployment

prefect deployment run "upstream/up-deploy"

What you should see - The worker picks up the upstream run and completes it. - The automation detects

prefect.flow-run.Completed

for the upstream deployment and triggers the downstream deployment. - The worker then runs the downstream deployment automatically. Optional: watch events for confirmation

prefect events stream --format json | jq 'select(.event | startswith("prefect.automation") or startswith("prefect.flow-run")) | {event, resource, related, occurred}'

If downstream doesn’t trigger, check: - Automation is enabled:

prefect automation ls

- Worker is connected to the same API as the server and polling the “local” pool - Events are visible:

prefect events stream --format json

- The automation’s

match_related

has the exact upstream deployment id printed by setup.py Want me to tailor this to your existing pool names or add parameters passed from upstream to downstream?