https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I ran into the situation where my pre...
# ask-marvin
m
@Marvin I ran into the situation where my prefect runner kubernetes pod was OOMkilled (killed because it ran out of memory). In such a scenario it is NOT able to send a status update and the workflow run keeps hanging in "Running" status. Even when I timeout is configured, the run doesn't seem to transition into the "Time out" status unless some action is taken on the run. After manually cancelling the run, it hangs on "Cancelling". What is needed for the run to actually go into the "Cancelled" status?
m
✅ thought for 213.2 seconds
What you’re seeing is generally expected when the execution infrastructure disappears before it can report a terminal state. Short version: - If the Kubernetes pod is OOMKilled, the flow code usually never gets a chance to send a final state update. - So the run can remain Running until something else detects the infrastructure failure and proposes a terminal state. - When you cancel a running flow, it first goes to Cancelling. - It only becomes Cancelled when a worker/runner-side process actually performs the cancellation cleanup and then submits the final 
Cancelled
state. So, to answer your direct question: What is needed for a run to actually go into 
Cancelled
? A Prefect worker/infrastructure manager must observe the run in 
Cancelling
, attempt to terminate the infrastructure (or determine it is already gone), and then successfully propose the 
Cancelled
state back to the API. If that process never happens, the run can remain stuck in Cancelling. --- ### Why this happens with OOMKilled Kubernetes pods An OOMKill is abrupt. The process inside the pod does not shut down cleanly, so it often cannot: - catch the event, - run cleanup logic, - send a final `Failed`/`Crashed`/`TimedOut` state. That means Prefect still thinks the run is active unless an external component notices the pod/job is gone or failed. In Kubernetes-based execution, that “external component” is typically the worker plus Kubernetes job observation logic. If that observation/cancellation path is interrupted or missing, the run can remain 
Running
. --- ### Why timeout may not transition the run A timeout in Prefect is not magic by itself if the underlying infrastructure has already vanished or stopped heartbeating in a way that prevents enforcement. In practice: - the pod dies, - no final state is reported, - the run can stay 
Running
, - and the timeout won’t necessarily immediately move it to a terminal state unless another orchestration/infrastructure actor evaluates and applies that transition. So your observation that it doesn’t move to timed out “unless some action is taken” is consistent with the infrastructure disappearing before Prefect can finish orchestration around the run. --- ### Why manual cancel can get stuck on 
Cancelling
Cancelling
is an intermediate state. It means:
“Prefect has accepted the cancellation request, and now some worker/infrastructure process needs to terminate the run infrastructure and finalize the state.”
For it to become 
Cancelled
, one of these must happen: 1. The worker finds the infrastructure and kills it, then marks the run 
Cancelled
. 2. The worker checks and determines the infrastructure is already gone, then marks the run 
Cancelled
. If no worker is available to do that, or if there is a bug/error in that path, the run can remain 
Cancelling
. There has also been at least one reported issue around Kubernetes jobs getting stuck in 
Cancelling
in certain 
prefect-kubernetes
versions: - GitHub issue #16001 --- ### In your exact scenario For a pod that was OOMKilled: - If the Kubernetes worker/observer detects the job failure cleanly, the run should generally move to a terminal state such as Crashed rather than hang forever. - If you manually cancel after the pod is already gone, the worker still needs to reconcile that fact and submit 
Cancelled
. - If there is no active worker processing that cancellation, or the worker cannot reconcile the infrastructure PID/job state, the run may sit in 
Cancelling
. --- ### What to check 1. Make sure a worker is running and healthy - The worker is what processes cancellation requests and infrastructure cleanup. 2. Check worker logs around the time of cancellation - Look for errors related to: - killing infrastructure - missing infrastructure PID - Kubernetes API errors - permissions deleting jobs/pods
3. Check the Kubernetes job/pod state - Was the job still present? - Was it deleted by TTL cleanup? - Did RBAC prevent Prefect from reading/deleting it? 4. Check your Prefect and 
prefect-kubernetes
versions - Especially if you’re seeing 
Cancelling
get stuck. - The issue above may be relevant. 5. Review memory limits / requests - If the pod is being OOMKilled, increasing memory limits or reducing workload memory pressure will prevent the root cause. --- ### CLI reference To cancel a flow run by ID, the CLI is:
Copy code
bash
prefect flow-run cancel <FLOW_RUN_ID>
Verified via CLI help. --- ### Practical expectation If the run infrastructure is already dead, a healthy worker should still be able to notice that and finalize the run. So if it stays in 
Cancelling
, that usually means one of: - no worker is available to process it, - the worker cannot access/reconcile the Kubernetes job, - or you’ve hit a bug/version-specific issue. --- ### Recommended next steps Please check and share: - your Prefect version - your 
prefect-kubernetes
version - whether you are using a Kubernetes work pool / worker or something else - worker logs from when the run entered 
Cancelling
If you want, I can help you narrow this down to: - expected behavior due to lost infrastructure, or - a specific stuck-cancellation bug/version issue.
m
@Marvin
A timeout in Prefect is not magic by itself if the underlying infrastructure has already vanished or stopped heartbeating in a way that prevents enforcement.
I understand that work is needed to transition from 
running
to 
time out
state, though I assumed this would be covered by the prefect API. Isn't that the whole point of having central orchestrator; to orchestrate?
The worker is what processes cancellation requests and infrastructure cleanup.
So you need a worker (not just a runner) to perform those cleanups? If so, this seems like a pretty serious issue with simplistic setup of just having a runner instead of a full worker, which you mentioned in another thread is a valid approach?
@Marvin we're running a minimalistic setup as explained in this thread; we're manually creating a 
Runner
, adding a flow to it using 
add_flow
and starting it using its 
start
method. From what I understood, this means, we only have a runner and we don't have a worker. Is that correct? From your previous response I get the impression that this is the reason why tasks stay on the "running" and "cancelling" status. Is this also correct? If the above is correct, what would be the simplest way to create a worker? We'd like to stay with our current minimal approach for our runners, because we we tried to 
serve
flows using the documenten methods (using the 
serve
method), it would do things like "downloading" flow implementations, which caused other complications. So we want a simple runner (like we have right now), but also a worker for handling the status changes.
m
✅ thought for 143.6 seconds
Yes — your understanding is correct. ### 1) Are you using only a runner and not a worker? Yes. If you are manually doing something like:
Copy code
python
runner = Runner(...)
runner.add_flow(...)
await runner.start()
then you are using a Runner-based setup, not a Worker-based setup. In that mode: - you do not have a worker - you do not have a work pool involved - the runner is responsible for polling, launching, and reconciling those runs So if your current execution model is just 
Runner + add_flow + start
, there is no separate worker in the architecture. --- ### 2) Is that why runs stay in 
Running
/ 
Cancelling
? Very likely yes, especially for the way you’re using Kubernetes. The important distinction is: - Runners are meant to manage the lifecycle of runs they launch - Workers are the component designed for infrastructure-backed execution via work pools, including Kubernetes job lifecycle handling If your flow execution pod is OOMKilled and disappears before reporting a final state, then some external orchestration component needs to notice that and reconcile the Prefect state. In a worker/work-pool Kubernetes setup, that role is handled by the worker + Kubernetes infrastructure handling. In your custom minimal runner setup, if the runner process does not have a robust mechanism to observe that Kubernetes infrastructure failure and finalize the run, then runs can remain: - 
Running
if nothing proposes a terminal state - 
Cancelling
if cancellation was requested but nothing completes the cancellation handshake So yes: the absence of a worker/infrastructure reconciliation path is a plausible reason for the stuck states you’re seeing. --- ### 3) Can you “just add a worker” to your current runner-managed flows? Not really. Workers and runners are different execution models. A worker works with: - deployments - work pools - infrastructure-specific job submission A runner works with: - flows added directly with 
add_flow
- runner-managed polling/execution So there is not really a supported “runner + separate worker for cleanup only” architecture. If a flow run is created and managed in runner mode, you cannot just attach a worker to it later and expect the worker to reconcile it. --- ### 4) Simplest way to get worker behavior while staying minimal The simplest supported path is: - move from manual runner-managed flows to deployment + work pool + worker - keep your code packaging simple - avoid 
from_source(...)
/ remote code pulling if that caused the “downloading” behavior you disliked In other words: - don’t use 
serve()
if that model caused problems for you - don’t use remote source-based deployments if you don’t want code download/pull behavior - instead use 
flow.deploy(...)
 with a Kubernetes work pool and an image that already contains your flow code That gives you: - Kubernetes-native execution - worker-managed cancellation - better reconciliation when pods die - no source download step at runtime --- ### 5) Recommended minimal architecture #### A. Put your flow code into a container image Build your own image containing: - your flow code - your dependencies - Prefect #### B. Create a Kubernetes work pool You can verify the CLI like this:
Copy code
bash
prefect work-pool --help
and then create a pool. Before suggesting the exact command, I should verify it if you want, but conceptually it’s a Kubernetes work pool. #### C. Deploy the flow to that work pool Example pattern in Prefect 3:
Copy code
python
from prefect import flow

@flow
def my_flow():
    print("hello")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-flow",
        work_pool_name="my-k8s-pool",
        image="my-registry/my-image:latest",
        build=False,
        push=False,
    )
This assumes the image already exists and already contains the code. #### D. Run a worker for that pool
The worker will poll the pool, create Kubernetes jobs, and handle cancellation/status reconciliation. --- ### 6) Important note about 
serve()
and “downloading code” What you described about “downloading” flow implementations is usually associated with remote source/storage patterns, not plain local 
serve()
itself. There are two different ideas here: #### Plain local serve
Copy code
python
my_flow.serve(...)
This is runner-based and runs from local code. No remote code pull is inherent in that by itself. #### Source-backed flow
Copy code
python
flow.from_source(...).serve(...)
This does involve pulling/downloading code from the configured source. So if your earlier setup was pulling code, I suspect it may have involved a source-backed pattern rather than plain local 
serve()
. --- ### 7) Bottom line Your conclusions are basically right: - Yes, your current setup is runner-only, not worker-based. - Yes, that is likely related to why runs can stay stuck in 
Running
or 
Cancelling
when Kubernetes infrastructure dies unexpectedly. - No, there isn’t really a simple supported way to bolt a worker onto runner-managed flows just for status handling. - Yes, the simplest robust alternative is to switch those flows to deployment + Kubernetes work pool + worker, while keeping things minimal by packaging code directly into the image and avoiding source download patterns. --- ### 8) What I’d recommend for you If your main goals are: - minimal setup - no remote source downloads - robust Kubernetes cancellation/state handling then use: - 
flow.deploy(...)
- a prebuilt Docker image containing your code - a Kubernetes work pool - a Kubernetes worker That is the smallest supported architecture that gives you proper worker-based reconciliation. --- ### Links - Prefect deployment concepts: https://docs.prefect.io/ - Workers and work pools: https://docs.prefect.io/ - Relevant stuck-cancellation issue: GitHub issue #16001 --- If you want, I can give you a minimal end-to-end example for: 1. creating a Kubernetes work pool 2. deploying your existing flow with 
flow.deploy(...)
3. starting the worker 4. doing it in a way that does not use remote source downloads.
m
@Marvin Thanks for those confirmations and recommendations. I'd just like to check one more thing; you mention that Runners are responsible for managing lifecycle runs they launch. Obviously a runner's pod getting OOMKilled is causing problems with that in my situations, but is it possible to have the runner, after it is restarted, manage the stale states (running/cancelling) instead of a my previous suggestion of creating a worker for that?
m
✅ thought for 131.5 seconds
Short answer: no, not by default. A restarted 
Runner
does not generally “re-adopt” old in-flight runs and reconcile stale 
Running
/ 
Cancelling
states from before it crashed. ### Why In runner mode, the runner tracks the runs/processes it launched in its own runtime state. If the runner pod gets OOMKilled: - that in-memory tracking is lost - on restart, the runner starts fresh - it polls for new scheduled work - it does not automatically reconstruct ownership of previously running/cancelling runs So if a run was left in: - 
Running
- 
Cancelling
because the runner died mid-lifecycle, the restarted runner typically will not clean that up automatically just by coming back. --- ### So is there a supported “runner restart recovery” path? Not really in the sense you’re asking. There is no general built-in orphan adoption mechanism for runner-managed runs where the restarted runner says:
“these stale runs used to be mine; I’ll now inspect them and finalize them.”
That’s why stale states can persist in this setup. --- ### What can you do instead? If you want to stay with runners, the best-supported approach is to add server-side stale run detection, not a worker. The main pattern is: - enable flow heartbeats - create an automation that marks runs as 
Crashed
if heartbeats stop arriving That way, if the runner dies and the run is stranded in 
Running
, Prefect can eventually mark it terminal without needing the original runner to come back and finish cleanup. Prefect documents this as zombie-flow detection: - Detect zombie flows --- ### Important nuance: 
Running
vs 
Cancelling
This heartbeat/automation approach is best for stale Running runs. For stale Cancelling runs, things are trickier: - 
Cancelling
implies cleanup was supposed to happen - if the runner that was supposed to perform that cleanup is gone, there is no built-in restart recovery that reliably resumes that cancellation process So for runner-only setups: - stale 
Running
can often be handled with heartbeat-based crash detection - stale 
Cancelling
is harder and may require manual or custom reconciliation --- ### Practical answer to your question
is it possible to have the runner, after it is restarted, manage the stale states instead of creating a worker?
Not in a built-in, reliable way. If you stay with runners, your realistic options are: 1. Use heartbeats + automation to mark stale 
Running
runs as 
Crashed
2. Build your own reconciliation script that periodically finds old 
Running
/ 
Cancelling
runs and force-updates them via the API 3. Move to worker/work-pool architecture if you want more robust infrastructure lifecycle reconciliation --- ### What I’d recommend for your case If you want to preserve the minimal runner approach, then the simplest improvement is: - keep the runner - add heartbeat-based zombie detection automation - optionally add a small periodic cleanup job for stale 
Cancelling
runs If you want, I can help you with either of these next: 1. a minimal zombie-flow automation for your runner setup 2. a small reconciliation script that finds stale 
Running
/ 
Cancelling
runs and marks them terminal 3. a comparison of those options vs switching to a worker-based Kubernetes setup
m
@Marvin it looks like registering the zombie flows automation works for automatically transitioning a flow to Crashed when there's no heartbeat anymore. I also have some application-specific state that I would also like to transition to the failed status in this scenario. Is it possible to trigger another flow in the same automation that can transition my application state? And is it possible to make this automation dependent on the type of flow-run that cashed?
m
✅ thought for 143.9 seconds
Yes — both are possible, with a couple of caveats. Short answer - Yes, a zombie-flow automation can do more than just mark the flow run `Crashed`; it can also trigger another action, including running another deployment/flow to update your app-specific state. - Yes, you can scope the automation so it only applies to certain flow runs, typically by deployment, tags, work pool, etc. - The main caveat is that filtering is based on event/resource metadata, not arbitrary Python-side “flow types” unless that distinction is represented in deployment/tags/work pool/resource labels. How to do the extra state transition A single automation can have multiple actions. For example: 1. 
change-flow-run-state
→ mark the zombie flow run 
Crashed
2. 
run-deployment
→ trigger a cleanup/reconciliation flow that updates your application state 3. optionally 
send-notification
or 
call-webhook
So your pattern would be: - Zombie detected (missing heartbeat) - Automation fires - Action 1 crashes the Prefect flow run - Action 2 runs a separate deployment that marks your application object/job/entity as failed This is the relevant docs area: - Detect zombie flows - Automations - Event triggers - Creating automations Important caveat about sequencing Automations support multiple actions, but they are not really “A then B with dependency semantics.” In practice: - you can attach both actions to the same automation - you should not assume strict transactional behavior between them If your app-state update must happen with special logic, the cleanest approach is often: - use the automation to mark the flow 
Crashed
- have the automation also run a dedicated cleanup/reconciliation deployment - put all your application-specific failure handling in that cleanup flow That gives you a clearer place to implement retries, idempotency, auditing, etc. Can you make it depend on the “type” of flow run that crashed? Usually yes, if that type is represented in Prefect metadata. Common ways to scope it: - specific deployment - specific tag(s) - specific work pool / work queue - sometimes related flow/deployment resources This is supported through automation event matching/filtering on the flow run event and its related resources: - Flow run events Practical recommendation If your “type” is application-specific, the easiest option is usually to encode it as a deployment or tag, then filter the automation on that. For example: - tag certain runs/deployments with 
app-type-x
- create the zombie automation to only match runs related to 
prefect.tag.app-type-x
What this likely looks like conceptually You’d configure one zombie automation with: - a proactive trigger for missing heartbeat - a filter for only the runs you care about - two actions: - 
ChangeFlowRunState(...CRASHED...)
- 
RunDeployment(...your cleanup deployment...)
If you want, I can help you write: 1. a UI-based setup for this automation, or 2. a Python example for an automation that: - crashes zombie runs - triggers another deployment - only applies to a chosen deployment/tag/work pool If you share how you distinguish your “flow-run type” today (deployment name, tags, parameters, work pool, etc.), I can suggest the exact filter to use.
m
@Marvin I would be great if I could have an automation based on flow name that could trigger a flow or deployment using just their name..
m
✅ thought for 124.6 seconds
Yes — partially. What you can do today 1. Filter an automation by flow name For flow-run events, the flow itself is a related resource, so you can scope the automation to a specific flow name. 2. Trigger another deployment/flow from the automation Yes, via the 
RunDeployment
action. What you can’t do directly - You cannot target a deployment “just by name” in 
RunDeployment
. - 
RunDeployment
requires a deployment ID when you explicitly choose the target deployment. So the answer is: - flow name filter by name? → Yes - run deployment by deployment name only? → No, not directly; use deployment ID Why For flow-run events, Prefect includes: - the flow run as the primary resource - the flow as a related resource, including 
prefect.resource.name
So automations can match on the related flow’s name. But for 
RunDeployment
, the action is designed to target a deployment by UUID, not by deployment name string. Docs - Automations - Event triggers - Flow run events - Detect zombie flows Practical pattern I’d recommend If you want “when flow named X becomes zombie, run cleanup Y”: - filter the automation by flow name - configure the action to run a specific deployment ID for cleanup/reconciliation So operationally it still feels name-based on the trigger side, but the action target is pinned to a concrete deployment. Example shape in Python Something like:
Copy code
trigger:
  match prefect.flow-run.*
  match_related:
    prefect.resource.role: flow
    prefect.resource.name: my-flow-name

actions:
  - change flow run state to Crashed
  - run deployment with deployment_id=<cleanup deployment uuid>
One more nuance If by “trigger a flow using just their name” you mean “dynamically choose which deployment to run based on the crashed flow’s name,” Prefect automations don’t provide a native “lookup deployment by name at execution time” action. In that case, common options are: - create one automation per flow name and point each to the right deployment ID - or trigger a single router/reconciliation deployment that receives 
{{ flow.name }}
/ 
{{ flow_run.id }}
and performs whatever app-specific logic you want That second option is often the cleanest. Good news on templating Action parameters can use templates, so your reconciliation deployment can receive context like: - 
{{ flow.name }}
- 
{{ flow_run.id }}
- 
{{ flow_run.parameters[...] }}
That makes it easy to drive your downstream state update logic. If you want, I can draft: - a UI configuration for this zombie-flow automation, or - a full Python example that filters by flow name and triggers a cleanup deployment with templated parameters.
m
@Marvin As I told you we're running a fairly minimal runner setup by using 
Runner.add_flow
and 
Runner.run
, so we don't have a worker. How would I trigger a flow that is being served this way in an automation?
m
✅ thought for 146.2 seconds
Yes — you still trigger it via a deployment. Even with a minimal 
Runner.add_flow(...)
/ 
Runner.run()
setup and no worker, the served flow is still backed by a deployment. That’s the key piece that makes automations work. How it works When you do something like:
Copy code
runner.add_flow(my_flow, name="my-deployment")
runner.run()
Prefect registers a deployment for that flow. Then: 1. the automation’s 
run-deployment
action creates a scheduled flow run for that deployment 2. your Runner polls for scheduled runs for the deployments it is serving 3. the Runner executes that run locally So: - worker required? → No - deployment required? → Yes - can Runner-served flows be triggered by automations? → Yes Important implication If you want an automation to trigger one of these served flows, that flow must be: - added to the same runner process (or another running runner) - registered as a deployment - actively being polled by that runner What you cannot do An automation cannot say “call this in-memory Python function by name on that runner.” It always works through Prefect’s orchestration layer, i.e. by creating a run for a deployment. So the pattern is - serve the target flow with 
Runner.add_flow(...)
- note the deployment it creates - configure automation action 
RunDeployment
to target that deployment - the Runner picks it up and runs it Your use case For your zombie-flow scenario, that likely means: - Flow A is running under your runner - zombie automation fires for Flow A - action 1 marks Flow A as 
Crashed
- action 2 runs deployment for Flow B, also served by a runner - Flow B updates your application-specific state to failed One caveat from your earlier question You said you’d like to trigger by name. The trigger side can be name-based enough, but the action side still wants a deployment ID, not just a deployment name. So in practice: - filter automation by flow name → yes - target served cleanup flow by deployment name only → no, use deployment ID Good news 
Runner.add_flow(...)
returns the deployment UUID, so you can capture it when registering the served flow. From the verified signature, 
Runner.add_flow(...)
returns a 
UUID
. So conceptually:
Copy code
cleanup_deployment_id = runner.add_flow(cleanup_flow, name="cleanup")
main_deployment_id = runner.add_flow(main_flow, name="main")
runner.run()
Then your automation can use 
cleanup_deployment_id
. Alternative that may be even cleaner Because 
Runner.add_flow
accepts 
triggers=...
, you may be able to define the event trigger directly on the served deployment itself, instead of creating it separately in the UI/API. That gives you a deployment-backed flow that is still served by the runner, but event-triggered. Relevant docs: - Deployments - Automations - Creating deployment triggers - Detect zombie flows Bottom line Yes — to trigger a flow served with 
Runner.add_flow
/ 
Runner.run
, configure the automation to run that flow’s deployment. The Runner will pick up the scheduled run and execute it locally; no worker is needed. If you want, I can sketch a concrete example showing: - a runner serving 
main_flow
and 
cleanup_flow
- a zombie automation filtered by flow name - the automation triggering the served 
cleanup_flow
deployment.
m
@Marvin I'm confused about the order of things, you suggest configuring the automation with the deployment-id. That means the automation has to be created when the deployment is created, which in my case happens every time I start a runner. I was under the impression that the automation would be created once and then it would never have to happen again. Could you suggest a flow with some example code that would create the deployment and the creation in a way that works for my scenario?
m
✅ thought for 106.8 seconds
Yes — your mental model is reasonable, and the good news is: you usually do not need a brand-new deployment ID every time the runner starts. With 
Runner.add_flow(...)
, Prefect uses a create-or-update by deployment name pattern. If you keep the same: - flow name - deployment name then on restart Prefect should update the existing deployment rather than create a completely new one, and the deployment ID remains stable. So for your scenario, the cleanest approach is: Recommended pattern 1. Start the runner 2. Call 
runner.add_flow(...)
for each served flow with stable names 3. Capture the returned deployment IDs 4. Create or update the automation programmatically 5. Run the runner That makes startup idempotent: - first startup creates deployments + automation - later startups update/reuse them You were right to push on the ordering question. Why this works 
Runner.add_flow(...)
does not just blindly create a new deployment every time. It looks up the deployment by full name and updates it if it already exists. So the automation can also be treated declaratively: - either create/update it on startup - or attach triggers to deployments where appropriate For your specific case, since you want: - one flow served by runner - another served flow triggered when zombie/crash condition happens - probably filtering by flow name I’d suggest an explicit startup function that: - registers both deployments - looks up/creates the automation that connects them Recommended architecture for your case - `main_flow`: your application flow - `reconcile_flow`: marks your application-specific state as failed - one startup/bootstrap script: - 
runner.add_flow(main_flow, name="main")
- 
runner.add_flow(reconcile_flow, name="reconcile")
- upsert automation: “when zombie/main_flow occurs, mark crashed and run reconcile deployment” - 
runner.run()
One important detail For filtering by flow name, use the event trigger filter on the related flow resource name. For triggering the reconciliation flow, use the deployment ID returned by 
add_flow
. Example Below is a concrete pattern you can adapt. ```python from datetime import timedelta from uuid import UUID from prefect import flow from prefect.runner import Runner from prefect.automations import Automation from prefect.events.schemas.automations import EventTrigger, Posture from prefect.events.schemas.events import ResourceSpecification from prefect.events.actions import ChangeFlowRunState, RunDeployment from prefect.client.schemas.objects import StateType AUTOMATION_NAME = "zombie-main-flow-reconcile" @flow def main_flow(app_id: str): print(f"Running main flow for {app_id}") # your real work here @flow def reconcile_flow(flow_run_id: str, flow_name: str, app_id: str | None = None): print( f"Reconciling app state for flow_run_id={flow_run_id}, " f"flow_name={flow_name}, app_id={app_id}" ) # update your application-specific state here def build_zombie_automation(reconcile_deployment_id: UUID, target_flow_name: str) -> Automation: return Automation( name=AUTOMATION_NAME, description="Crash zombie runs of main_flow and reconcile app state", 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", }, posture=Posture.Proactive, threshold=1, within=timedelta(seconds=90), match=ResourceSpecification( { "prefect.resource.id": ["prefect.flow-run.*"], } ), match_related=ResourceSpecification( { "prefect.resource.role": "flow",
"prefect.resource.name": target_flow_name, } ), for_each={"prefect.resource.id"}, ), actions=[ ChangeFlowRunState( state=StateType.CRASHED, message="Marked crashed after heartbeat stopped.", ), RunDeployment( source="selected", deployment_id=reconcile_deployment_id, parameters={ "flow_run_id": "{{ flow_run.id }}", "flow_name": "{{ flow.name }}", "app_id": "{{ flow_run.parameters.get('app_id') }}", }, ), ], ) async def upsert_automation(automation: Automation): existing = await Automation.aread(name=automation.name) if existing is None: await automation.acreate() print(f"Created automation: {automation.name}") else: automation.id = existing.id await automation.aupdate() print(f"Updated automation: {automation.name}") async def main(): runner = Runner() # Keep these names stable across restarts await runner.aadd_flow(main_flow, name="main") reconcile_deployment_id = await runner.aadd_flow(reconcile_flow, name="reconcile") automation = build_zombie_automation( reconcile_deployment_id=reconcile_deployment_id, target_flow_name="main_flow", # this is the Python flow name unless you've customized it ) await upsert_automation(automation) await runner.start() if name == "__main__": import asyncio asyncio.run(main())
Copy code
*What this gives you*
- deployment registration is stable across restarts
- automation is created once, then updated on later restarts
- no worker required
- reconciliation flow is also served by the same runner model

*Two things to double-check in your environment*

1. *Flow name vs deployment name*  
   The trigger filter above uses *flow name*, not deployment name.  
   If your flow is declared as:

   ```python
   @flow
   def main_flow(...):
       ...
then the flow resource name is typically 
main_flow
. 2. Automation read/update behavior The pattern above assumes 
Automation.aread(name=...)
returns either the existing automation or 
None
. If you want, I can give you a more defensive version that uses the client directly in case your installed version behaves slightly differently. Alternative: deployment-managed triggers There is another pattern: define 
triggers=
directly on the deployment created by 
Runner.add_flow(...)
. That is nice when the automation is tightly coupled to that deployment. But for your case, because the automation: - watches one flow - triggers another deployment - may include custom app-state logic I think the explicit startup upsert approach is clearer and easier to maintain. My recommendation Use a single bootstrap script that always: - registers deployments with stable names - upserts the automation by name - starts the runner That preserves the “declare once, keep it synced” experience you were expecting. If you want, I can tighten this into a fully verified Prefect 3 example for either: - pure async startup with 
Runner
, or - a simpler version using 
flow.serve(...)
instead of manually constructing 
Runner
.
m
@Marvin Thanks! The example code in your previous response was extremely helpful. One more question about it; is it possible to configure the 
RunDeployment
action to pass the all the arguments from the original flow (
main_flow
in your example) to the 
reconcile_flow
, as well as other (statically defined) arguments?
m
✅ thought for 106.6 seconds
Yes — partially. You can configure 
RunDeployment
to pass: 1. selected parameters from the triggering flow run, and 2. additional static values But you cannot automatically forward the entire original parameter dict in one shot. So this works conceptually:
Copy code
python
RunDeployment(
    source="selected",
    deployment_id=reconcile_deployment_id,
    parameters={
        "name": {
            "__prefect_kind": "jinja",
            "template": "{{ flow_run.parameters['name'] }}",
        },
        "date": {
            "__prefect_kind": "jinja",
            "template": "{{ flow_run.parameters['date'] }}",
        },
        "mode": "reconcile",
        "retries_allowed": 3,
    },
)
That means: - 
name
and 
date
come from `main_flow`'s run parameters - 
mode
and 
retries_allowed
are static values you define in the automation Important limitation: there isn't a built-in "spread all upstream parameters into this downstream deployment's parameters" feature for 
RunDeployment
. You need to map them explicitly. Also, if you provide 
parameters=...
, those are the parameters sent to the target deployment — it's not doing an automatic merge with the target deployment's defaults for you in the automation config. So if the downstream flow expects parameters, you should define them explicitly. A practical example:
Copy code
python
from prefect.automations import Automation, EventTrigger, Posture, RunDeployment
from datetime import timedelta

automation = Automation(
    name="Trigger reconcile flow",
    trigger=EventTrigger(
        expect={"prefect.flow-run.Completed"},
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(seconds=30),
    ),
    actions=[
        RunDeployment(
            source="selected",
            deployment_id=reconcile_deployment_id,
            parameters={
                "customer_id": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['customer_id'] }}",
                },
                "as_of_date": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.parameters['as_of_date'] }}",
                },
                "triggering_flow_run_id": {
                    "__prefect_kind": "jinja",
                    "template": "{{ flow_run.id }}",
                },
                "reconcile_mode": "full",
            },
        )
    ],
)
Docs: - Automations - Jinja templating in automations - Access parameters in templates If you want, I can also show you a pattern for making this feel like "pass everything through" — for example by designing 
reconcile_flow
to accept a single 
upstream_params
dict plus a few explicit static fields.
m
@Marvin is it required to specify both a 
template
and a 
"__prefect__kind"
(with 
_prefect_kind
set to 
"jinja"
) for each argument (as opposed to simply giving the template as a single value for each parameter)? You didn't do that in your previous example...
m
✅ thought for 12.3 seconds
Good catch — and no, it is not always required to specify both. For 
RunDeployment.parameters
, Prefect supports plain templated string values and the explicit structured form with 
__prefect_kind
. So these are both valid patterns: Simple form
Copy code
python
parameters={
    "customer_id": "{{ flow_run.parameters['customer_id'] }}",
    "mode": "full",
}
Explicit form
Copy code
python
parameters={
    "customer_id": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['customer_id'] }}",
    },
    "mode": "full",
}
The difference is mainly about how Prefect interprets the value and whether you need more control. Use the simple string form when: - the parameter is just a Jinja template string - you're okay with normal template rendering behavior - the rendered value being string-like is fine Use the explicit 
__prefect_kind: "jinja"
form when: - you want to be unambiguous - you're working with nested structures - you need schema/hydration behavior that Prefect's automation templating system understands explicitly - you may later combine this with other kinds like 
json
or 
workspace_variable
One important nuance: if you want to preserve non-string types cleanly, the explicit structured form is safer. Plain Jinja templating often behaves like string rendering unless you wrap it carefully. So for example, if 
count
should remain an integer, I would not rely blindly on the simple form:
Copy code
python
parameters={
    "count": {
        "__prefect_kind": "jinja",
        "template": "{{ flow_run.parameters['count'] }}",
    }
}
Depending on the downstream parameter typing, that may still be okay, but if type preservation matters, it's worth testing. So the short version is: - No, you do not have to provide both for every parameter. - Yes, a plain template string can work. - But the explicit 
__prefect_kind
form is the safer/recommended option when you need predictable templating behavior, especially for nested or typed values. If you'd like, I can rewrite the earlier example in the minimal valid syntax style so it's easier to copy/paste.
m
@Marvin I'm using the pattern you suggested, and I'm passing the following parameters into the 
RunDeployment
action:
Copy code
parameters={
                        "revision": {
                            "__prefect_kind": "jinja",
                            "template": "{{ flow_run.parameters['revision'] }}",
                        },
                        "retry": True,
                    },
But the prefect API is still giving the following error when it tries to scheduled the new flow:
Copy code
Action failed: "Validation error occurred for 'run-deployment' - Error creating flow run: Validation failed for field 'revision'. Failure reason: '0' is not of type 'integer'"
Do you have a suggestion for how to properly pass the revision argument from the original flow_run into the new flow_run (in this case I'm simply trying to reschedule the same flow with ther additional "retry" argument
m
✅ thought for 72.2 seconds
Yes — the issue is type preservation. Your current config:
Copy code
python
"revision": {
    "__prefect_kind": "jinja",
    "template": "{{ flow_run.parameters['revision'] }}",
}
renders 
revision
as a string, so if the original value is 
0
, Prefect ends up validating 
"0"
instead of 
0
. That matches the error:
Copy code
text
'0' is not of type 'integer'
The fix is to wrap the Jinja expression in a 
json
value and use `| tojson`:
Copy code
python
parameters={
    "revision": {
        "__prefect_kind": "json",
        "value": {
            "__prefect_kind": "jinja",
            "template": "{{ flow_run.parameters['revision'] | tojson }}",
        },
    },
    "retry": True,
}
That preserves the original type, so an integer stays an integer. You can also use the shorter form:
Copy code
python
parameters={
    "revision": "{{ flow_run.parameters['revision'] }}",
    "retry": True,
}
In Prefect 3.x, Prefect will automatically upgrade a single-expression template like that into the typed 
json + jinja + tojson
form internally. But since you're already writing the explicit structured version, you need to use the correct typed structure yourself. So the practical rule is: - 
__prefect_kind: "jinja"
→ good for string templating - 
__prefect_kind: "json"
wrapping a Jinja template with 
| tojson
→ good for preserving ints / bools / dicts / lists / nulls For your reschedule case, I'd use:
Copy code
python
parameters={
    "revision": {
        "__prefect_kind": "json",
        "value": {
            "__prefect_kind": "jinja",
            "template": "{{ flow_run.parameters['revision'] | tojson }}",
        },
    },
    "retry": True,
}
Relevant docs: - Jinja templating in automations - Access parameters in templates If you want, I can rewrite your full 
RunDeployment(...)
block in the safest form for rescheduling the same deployment with one modified parameter.
m
@Marvin I'm now creating an automation, almost exactly like you suggested, this is the python code that builds the aumation:
Copy code
return Automation(
        name=automation_name,
        description=f"Crash zombie runs of {target_flow_name} and trigger retry flow",
        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",
            },
            posture=Posture.Proactive,
            threshold=1,
            within=timedelta(seconds=90),
            match=ResourceSpecification({
                "prefect.resource.id": ["prefect.flow-run.*"],
            }),
            match_related=ResourceSpecification({
                "prefect.resource.role": "flow",
                "prefect.resource.name": target_flow_name,
            }),
            for_each={"prefect.resource.id"},
        ),
        actions=[
            ChangeFlowRunState(
                state=StateType.CRASHED,
                message="Marked crashed after heartbeat stopped.",
            ),
            RunDeployment(
                source="selected",
                deployment_id=deployment_id,
                parameters=parameters,
            ),
        ],
    )
No I have some workflow runs that have been transitioned to the "crashed" status and for which a new deployment/flow is scheduled, but I also still have some other runs for the same workflow that are still getting stuck on the "running" status, without any new activity, but they are not getting picked up by this automation. Do you know what could be the cause of this? Those flow runs that are still stuck in "running" did receive some heartbeats in the first minutes after they were started (as shown in the prefect server UI). One shows that it was active for 17 hours (one task also show being active for all that time) but during those 17 hours there was not activity and the prefect UI just show the unterminated "in progress" task with the timeline cursor frozen at start-time + 17h27m....
m
✅ thought for 210.2 seconds
Yes — the most likely cause is how this trigger behaves when 
heartbeat
is in both 
after
and 
expect
. In your automation:
Copy code
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",
}
posture=Posture.Proactive
within=90s
threshold=1
that means each matching heartbeat can effectively restart / extend the trigger window for that flow run. So if a run: - emits an initial heartbeat, - emits another heartbeat before the 90s window closes, - maybe emits a few more, - then gets stuck, the automation may not fire the way you expect because the “missing heartbeat” condition is being evaluated relative to the latest bucket/window, not simply “has this flow been inactive for 90 seconds since it first started?” That matches your symptom exactly: - some runs get marked 
Crashed
- others remain 
Running
forever even though they look dead in the UI A few likely contributors: 1. Heartbeats may have reset the proactive window If those stuck runs emitted heartbeats for the first few minutes, the trigger window may have kept getting pushed forward. Depending on event timing, some runs will age out cleanly and some won’t. 2. Event matching may not be identical for every heartbeat event You’re also filtering on:
Copy code
match_related=ResourceSpecification({
    "prefect.resource.role": "flow",
    "prefect.resource.name": target_flow_name,
})
If some heartbeat events for those runs are missing or differing in related-resource metadata, they may not match the automation consistently. 3. Per-run bucketing depends on 
prefect.resource.id
being present as expected You’re using:
Copy code
for_each={"prefect.resource.id"}
That is usually right, but if an event is missing that label or shaped unexpectedly, it can affect how the automation tracks that run. 4. Timing / ordering edge cases Proactive automations are evaluated on a schedule, so delayed or out-of-order heartbeat events can create inconsistent behavior between otherwise similar runs. The stuck “Running” + frozen task timeline part is also important: that suggests the engine/process got wedged badly enough that it stopped making meaningful progress, but not in a way that always produces a clean terminal event. In those cases, heartbeat-based detection is exactly the right idea — but this particular trigger configuration is a bit fragile. What I would change Instead of including heartbeat in both 
after
and 
expect
, I’d strongly suspect you want the automation to mean: - start watching after a heartbeat occurs - if no further expected event arrives within N seconds, fire But when 
heartbeat
is included in 
expect
, each heartbeat is itself treated as a satisfying/rolling event. So the first thing I would try is removing heartbeat from 
expect
and only using terminal events there. Something like:
Copy code
EventTrigger(
    after={"prefect.flow-run.heartbeat"},
    expect={
        "prefect.flow-run.Completed",
        "prefect.flow-run.Failed",
        "prefect.flow-run.Cancelled",
        "prefect.flow-run.Crashed",
    },
    posture=Posture.Proactive,
    threshold=1,
    within=timedelta(seconds=90),
    match=ResourceSpecification({
        "prefect.resource.id": ["prefect.flow-run.*"],
    }),
    match_related=ResourceSpecification({
        "prefect.resource.role": "flow",
        "prefect.resource.name": target_flow_name,
    }),
    for_each={"prefect.resource.id"},
)
That makes the logic more like: - heartbeat starts the watch window - if no terminal event happens within 90s, trigger the actions That said, there’s an important caveat: This only works if you actually want “no terminal event within 90s after heartbeat”, not strictly “no heartbeat for 90s.” Those are slightly different semantics. If your real goal is specifically:
“If heartbeats stop for 90 seconds while the run is still Running, mark it crashed” then Prefect’s event trigger semantics can be tricky here, and your current setup may hit edge cases. What I’d verify next For one affected run and one unaffected run, compare the emitted events: - are heartbeat events still arriving for the missed run? - do those heartbeat events include the same related 
flow
resource? - is 
prefect.resource.id
present as expected? - did the run emit any unusual state transition events? If you can inspect the event stream for a “stuck but missed” run, that will probably show why it didn’t match the automation. Useful docs / references - Automations concepts: docs.prefect.io/v3/concepts/automations - Event triggers: docs.prefect.io/v3/concepts/event-triggers If you want, I can help you rewrite this automation into a safer pattern for “zombie run detection,” or help you reason through the exact semantics of 
after
/ 
expect
/ 
threshold
for your use case.
m
@Marvin, yes please provide a suggested rewrite of the automation, And could you also answer one question about your current suggestion; if I remove the hearbeat from the "expect" and keep it only in the "after" section of the automation, will it then only trigger if there is no terminal event after the latest heartbeat (because it shouldn't trigger simply because a flow takes longer than 90 seconds to complete and there has not been a terminal event since the first hearbeat).
m
✅ thought for 24.5 seconds
Yes — with 
heartbeat
only in 
after
, the watch window is anchored to the latest matching 
after
event, not the first one. So if a flow keeps heartbeating normally, each new heartbeat should start a fresh 90s watch window. In that setup, the automation should only fire when there has been no subsequent qualifying event after the most recent heartbeat for that flow run. So the intent becomes: - heartbeat happens - open a 90s window for that flow run - if a terminal event arrives in that window, do nothing - if another heartbeat arrives, the window effectively rolls forward - if nothing arrives before the window expires, fire the automation That is much closer to “heartbeat stopped” semantics. Suggested rewrite I’d suggest this version first:
Copy code
from datetime import timedelta
from prefect.events.schemas.automations import Automation, EventTrigger, Posture
from prefect.events.schemas.events import ResourceSpecification
from prefect.automations import ChangeFlowRunState, RunDeployment
from prefect.client.schemas.objects import StateType

def build_zombie_flow_automation(
    automation_name: str,
    target_flow_name: str,
    deployment_id: str,
    parameters: dict,
) -> Automation:
    return Automation(
        name=automation_name,
        description=f"Crash zombie runs of {target_flow_name} and trigger retry flow",
        trigger=EventTrigger(
            after={"prefect.flow-run.heartbeat"},
            expect={
                "prefect.flow-run.Completed",
                "prefect.flow-run.Failed",
                "prefect.flow-run.Cancelled",
                "prefect.flow-run.Crashed",
            },
            posture=Posture.Proactive,
            threshold=1,
            within=timedelta(seconds=90),
            match=ResourceSpecification({
                "prefect.resource.id": ["prefect.flow-run.*"],
            }),
            match_related=ResourceSpecification({
                "prefect.resource.role": "flow",
                "prefect.resource.name": target_flow_name,
            }),
            for_each={"prefect.resource.id"},
        ),
        actions=[
            ChangeFlowRunState(
                state=StateType.CRASHED,
                message="Marked crashed after heartbeat stopped.",
            ),
            RunDeployment(
                source="selected",
                deployment_id=deployment_id,
                parameters=parameters,
            ),
        ],
    )
Why this is safer - 
heartbeat
starts or refreshes monitoring - only terminal events satisfy 
expect
- long-running healthy flows should not be crashed just because they run for a long time, as long as heartbeats keep arriving - once heartbeats stop, no new 
after
event refreshes the window, so after 90s the proactive trigger can fire Important caveat There is still one thing I’d be careful about: this automation is still fundamentally event-timing-based. So if: - heartbeat delivery is delayed, - some heartbeats don’t match your 
match_related
filter, - or events arrive out of order, you can still see occasional misses or false positives. Because of that, I’d recommend loosening the matching slightly unless you really need the flow-name filter. For example, if the deployment/run selection is already controlled elsewhere, this is more robust:
Copy code
trigger=EventTrigger(
    after={"prefect.flow-run.heartbeat"},
    expect={
        "prefect.flow-run.Completed",
        "prefect.flow-run.Failed",
        "prefect.flow-run.Cancelled",
        "prefect.flow-run.Crashed",
    },
    posture=Posture.Proactive,
    threshold=1,
    within=timedelta(seconds=90),
    match=ResourceSpecification({
        "prefect.resource.id": ["prefect.flow-run.*"],
    }),
    for_each={"prefect.resource.id"},
)
Then scope the action logic or deployment parameters instead of relying on 
match_related
if possible. One more concern with your action list You currently do both:
1. 
ChangeFlowRunState(... CRASHED ...)
2. 
RunDeployment(...)
That can work, but be aware: - if the original run is not actually recoverable, marking it 
CRASHED
is fine - but your retry deployment may launch even in edge cases where the original process is still alive but just badly stalled - that can create duplicate side effects unless your flow is idempotent So if the flow touches external systems, I’d strongly recommend ensuring: - idempotency keys / deduping - or a parameter passed to the retry flow that tells it which original flow-run it is replacing Example:
Copy code
RunDeployment(
    source="selected",
    deployment_id=deployment_id,
    parameters={
        **parameters,
        "original_flow_run_id": "{{ event.resource.id }}",
    },
)
If your templating supports that in your automation context, it can be very useful for tracing and dedupe. My practical recommendation I would test in this order: Option A — minimal fix Use the rewrite above and only remove 
heartbeat
from 
expect
. Option B — make matching less strict If misses continue, remove 
match_related
and rely only on: - 
prefect.resource.id
- maybe a broader flow filter if absolutely needed Option C — increase the timeout slightly If 90s is close to your heartbeat cadence or you sometimes have transient event delays, try 
within=timedelta(seconds=120)
or 
180
. Best next debugging step Take one missed run and inspect: - last heartbeat timestamp - whether any later heartbeat events existed - whether those events included the related flow resource and flow name exactly as expected - whether the run ever emitted a terminal event late If you want, I can also give you: 1. a more defensive rewrite of the automation, or 2. a small event-inspection script to compare a “caught” run and a “missed” run.
m
@Marvin you're right about the idempotency issue. I'm actually removing the 
RunDeployment
action for now and only transitioning flow runs to 
Crashed
. One more question; the 
for_each
argument to the 
EventTrigger
now get 
{"prefect.resource.id"}
, does this mean it will potentially trigger on other resources (steps?) besides flow-runs as well? Could I make it more specific to just apply to flow runs?
m
✅ thought for 12.1 seconds
for_each={"prefect.resource.id"}
does not broaden the trigger to other resource types by itself. 
for_each
just says: bucket / track matching events separately by this label. In your case, separately per resource id. What determines which events are eligible at all is mainly 
match
, 
match_related
, 
after
, and 
expect
. Since you already have:
Copy code
match=ResourceSpecification({
    "prefect.resource.id": ["prefect.flow-run.*"],
})
that is the part making it specific to flow-run resources. So no, 
for_each={"prefect.resource.id"}
does not mean “also steps/tasks/etc”. It means: - only events whose primary resource id matches 
prefect.flow-run.*
- and then among those, evaluate each distinct flow-run id independently So this is already the right way to scope it to flow runs. Recommended version If you’re removing the retry action, I’d keep it like this:
Copy code
Automation(
    name=automation_name,
    description=f"Mark zombie runs of {target_flow_name} as crashed",
    trigger=EventTrigger(
        after={"prefect.flow-run.heartbeat"},
        expect={
            "prefect.flow-run.Completed",
            "prefect.flow-run.Failed",
            "prefect.flow-run.Cancelled",
            "prefect.flow-run.Crashed",
        },
        posture=Posture.Proactive,
        threshold=1,
        within=timedelta(seconds=90),
        match=ResourceSpecification({
            "prefect.resource.id": ["prefect.flow-run.*"],
        }),
        match_related=ResourceSpecification({
            "prefect.resource.role": "flow",
            "prefect.resource.name": target_flow_name,
        }),
        for_each={"prefect.resource.id"},
    ),
    actions=[
        ChangeFlowRunState(
            state=StateType.CRASHED,
            message="Marked crashed after heartbeat stopped.",
        ),
    ],
)
If you want to make it even more clearly flow-run specific You already mostly have. The strongest scoping signal is:
Copy code
"prefect.resource.id": ["prefect.flow-run.*"]
That’s the important one. If you want, you can also think about whether 
match_related
is helping or hurting: - helping if you truly only want one flow name - hurting if some heartbeat events don’t consistently include that related flow metadata So if you still see missed runs, I’d actually suspect 
match_related
before I’d suspect 
for_each
. Short answer - 
for_each
does not expand the trigger to other resource types - your 
match
filter is what keeps it limited to flow runs - 
for_each={"prefect.resource.id"}
is the correct per-flow-run bucketing choice If you want, I can also suggest a slightly simplified final automation definition optimized for “mark stalled flow runs as crashed” with the least fragile matching.
m
@Marvin Please do, though please first answer one more question; with the more flexible matching approach will it then also consider heartbeats that are being given for another flow-run of the same flow?
m
✅ thought for 19.4 seconds
No — if you keep 
for_each={"prefect.resource.id"}
, heartbeats from one flow run should not satisfy or refresh the trigger window for a different flow run. That’s the key separation mechanism. So even with more flexible matching, the logic is still effectively: - collect matching events - group them by 
prefect.resource.id
- evaluate each resource id independently Since a flow run’s resource id is unique, one run’s heartbeat should not count for another run. Example: - run A has resource id 
prefect.flow-run.111
- run B has resource id 
prefect.flow-run.222
With:
Copy code
match=ResourceSpecification({
    "prefect.resource.id": ["prefect.flow-run.*"],
}),
for_each={"prefect.resource.id"},
Prefect should keep separate buckets for: - 
prefect.flow-run.111
- 
prefect.flow-run.222
So a heartbeat from run B should not prevent run A from being marked crashed. Why I suggested more flexible matching The thing I’d relax is not the per-run separation. I’d relax the flow-name-related-resource filter, because that is more likely to exclude some legitimate heartbeat events unexpectedly. So instead of:
Copy code
match_related=ResourceSpecification({
    "prefect.resource.role": "flow",
    "prefect.resource.name": target_flow_name,
})
I’d try relying on: - 
match
to limit to 
prefect.flow-run.*
- 
for_each
to isolate each flow run That makes the automation less brittle while still keeping runs independent. Suggested simplified version
Copy code
from datetime import timedelta

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


def build_zombie_crash_automation(automation_name: str) -> Automation:
    return Automation(
        name=automation_name,
        description="Mark flow runs as crashed when heartbeats stop",
        trigger=EventTrigger(
            after={"prefect.flow-run.heartbeat"},
            expect={
                "prefect.flow-run.Completed",
                "prefect.flow-run.Failed",
                "prefect.flow-run.Cancelled",
                "prefect.flow-run.Crashed",
            },
            posture=Posture.Proactive,
            threshold=1,
            within=timedelta(seconds=120),
            match=ResourceSpecification({
                "prefect.resource.id": ["prefect.flow-run.*"],
            }),
            for_each={"prefect.resource.id"},
        ),
        actions=[
            ChangeFlowRunState(
                state=StateType.CRASHED,
                message="Marked crashed after heartbeat stopped.",
            ),
        ],
    )
Why this is probably the least fragile version - only targets 
flow-run
resources - buckets independently per flow run - doesn’t depend on related-resource metadata being present/consistent - treats heartbeat as the “keepalive” signal - terminal states suppress the crash action - slightly larger 
within
window (
120s
) gives more tolerance for event delay/jitter One caution If you remove the flow-name filter entirely, this automation applies to all flow runs whose heartbeat events it sees, not just one specific flow. So the simplified version is best if: - that is what you want, or - you can scope the automation some other reliable way If you do need it to apply to only one flow, I’d prefer filtering by a more reliable resource attribute if one is consistently present on the event stream for your environment. But between the two, I would trust: - 
prefect.resource.id
on the primary resource more than - 
match_related
on related flow metadata Direct answer - No, flexible matching does not mean one run’s heartbeat keeps another run alive - 
for_each={"prefect.resource.id"}
keeps them separate - relaxing 
match_related
only makes event inclusion broader, not bucket sharing broader
If you want, I can give you two final versions next: 1. a single-flow version, and 2. an all-flow-runs version, with notes on when to choose each.
m
@Marvin I've tried your new automation suggestion (see my code below; I'm not passing in a 
retry_delpoyment_id
or 
retry_parameters
at this moment), but now the automation never kicks in and the zombie flow stay in "running" status again indefinitely, despite not receiving any heartbeats.
Copy code
def _build_zombie_crash_automation(
    target_flow_name: str,
    retry_deployment_id: UUID | None = None,
    retry_parameters: dict[str, Any] | None = None,
) -> Automation:
    """
    Builds a prefect automation instance that would trigger two actions when a flow_run for the flow
    with name=target_flow_name stops receiving a hearbeat, while still in the `running` status:
    - Action 1: It's status will be updated to `Crashed`
    - Action 2 (optional): The same flow (identified by `retry_deployment_id`) will be scheduled with the given parameters
    """
    return Automation(
        name=f"zombie_{target_flow_name}_to_crashed_and_reschedule" if retry_deployment_id else f"zombie_{target_flow_name}_to_crashed",
        description=f"Transition zombie runs of {target_flow_name} to 'Crashed'"
        + (" and trigger retry flow" if retry_deployment_id else ""),
        trigger=EventTrigger(
            # Within 2 minutes (regular heatbeat interval is 30 seconds)...
            within=timedelta(seconds=120),
            # ...after the _last_ received heartbeat...
            after={"prefect.flow-run.heartbeat"},
            # ...expect a terminal event
            expect={
                "prefect.flow-run.Completed",
                "prefect.flow-run.Failed",
                "prefect.flow-run.Cancelled",
                "prefect.flow-run.Crashed",
            },
            posture=Posture.Proactive,
            threshold=1,
            match=ResourceSpecification({
                "prefect.resource.id": ["prefect.flow-run.*"],
            }),
            for_each={"prefect.resource.id"},
        ),
        actions=[
            ChangeFlowRunState(
                state=StateType.CRASHED,
                message="Marked crashed after heartbeat stopped.",
            ),
            *(
                [
                    RunDeployment(
                        source="selected",
                        deployment_id=retry_deployment_id,
                        parameters=retry_parameters or {},
                    ),
                ]
                if retry_deployment_id
                else []
            ),
        ],
    )
m
✅ thought for 257.4 seconds
Yes — I found the key issue. Your new version likely never fires because removing 
heartbeat
from 
expect
also removes the “sliding window from the latest heartbeat” behavior you actually need. What’s happening With:
Copy code
after={"prefect.flow-run.heartbeat"}
expect={
    "prefect.flow-run.Completed",
    "prefect.flow-run.Failed",
    "prefect.flow-run.Cancelled",
    "prefect.flow-run.Crashed",
}
posture=Posture.Proactive
threshold=1
within=120s
for_each={"prefect.resource.id"}
the first heartbeat creates the bucket, but later heartbeats do not reset it the way you want, because the special proactive reset behavior only happens when an event is in both 
after
and 
expect
. So the configuration that best matches “fire 120s after the latest heartbeat if no more flow-run events arrive” is actually closer to:
Copy code
expect={"prefect.flow-run.*"}
not just terminal events. That sounds counterintuitive, but it aligns with Prefect’s event trigger behavior. *Why 
expect={"prefect.flow-run.*"}
works better* That causes: - heartbeat to count as an expected event - each new heartbeat to roll the window forward - terminal events to also count - trigger to fire only when no matching flow-run event arrives within the window So the watch becomes: - heartbeat at t=0 - another heartbeat at t=30s -> reset window - another at t=60s -> reset window - then silence - fire around t=180s That is much closer to “heartbeat stopped after the latest heartbeat”. Suggested final automation Since you’re only marking the original run as 
Crashed
, I’d suggest this:
Copy code
from datetime import timedelta
from typing import Any
from uuid import UUID

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


def _build_zombie_crash_automation(
    target_flow_name: str,
) -> Automation:
    return Automation(
        name=f"zombie_{target_flow_name}_to_crashed",
        description=f"Transition zombie runs of {target_flow_name} to Crashed",
        trigger=EventTrigger(
            within=timedelta(seconds=120),
            after={"prefect.flow-run.heartbeat"},
            expect={"prefect.flow-run.*"},
            posture=Posture.Proactive,
            threshold=1,
            match=ResourceSpecification({
                "prefect.resource.id": ["prefect.flow-run.*"],
            }),
            for_each={"prefect.resource.id"},
        ),
        actions=[
            ChangeFlowRunState(
                state=StateType.CRASHED,
                message="Marked crashed after heartbeat stopped.",
            ),
        ],
    )
Why this version is better - scoped to flow runs via 
match
- isolated per flow run via 
for_each
- rolling window based on latest heartbeat - doesn’t depend on 
match_related
, which may be brittle - doesn’t try to launch a replacement run Important caveat With 
expect={"prefect.flow-run.*"}
, any flow-run event for that same run can refresh/satisfy the bucket, not just heartbeats. Usually that’s okay, because: - terminal events are fine - state-transition events are still legitimate evidence the run is active/changing But it means the semantics become: “mark the run crashed if no flow-run events at all happen for 120s after the latest heartbeat/event cycle” not strictly: “mark crashed if no heartbeat arrives for 120s.” In practice, that may actually be preferable for zombie detection. Answering your concern about other flow runs of the same flow No — not if you keep:
Copy code
for_each={"prefect.resource.id"}
That still isolates evaluation per unique flow-run id, so one run’s events should not keep another run alive. One more subtle thing
You may still see the automation fire slightly after 120 seconds, not exactly at 120 seconds. There can be a small scheduler/evaluation delay. So think of it as: - expected around 120s+ - not guaranteed at exactly 120.000s If you want to keep it limited to one specific flow Then the problem becomes filtering reliably without reintroducing brittleness. The 
match_related
approach may be what was causing misses earlier. If you need single-flow scoping, I’d suggest testing the broad version above first to confirm the trigger logic works reliably. If it does, then add flow-specific filtering carefully. Docs - Automations: docs.prefect.io/v3/concepts/automations - Event triggers: docs.prefect.io/v3/concepts/event-triggers If you want, I can next help you with one of these: 1. a single-flow-only version with the least fragile filter possible, or 2. a small validation checklist to confirm the event stream for one zombie run matches what this automation expects.
m
@Marvin thanks for the suggestion, but you initially recommended explicitly to remove the heartbeat from the Trigger's 
expect
argument, and now you're suggesting to include anything related to the flow-run in the 
expect
argument. This seems rather contradictory. Could you review your latest suggestion one more time to double-check that you're not overlooking side-effects?
m
✅ thought for 25.1 seconds
You’re right to call that out — my earlier recommendation to remove 
heartbeat
from 
expect
was not the best advice for your goal. Thanks for pushing on it. Corrected view If your goal is specifically: “mark a flow run as 
Crashed
when heartbeats stop for N seconds” then removing 
heartbeat
from 
expect
is not the right fit, because it breaks the “window follows the latest heartbeat” behavior. So yes: my two suggestions were in tension, and the later one is the one I trust more after reviewing the trigger behavior more carefully. That said, your concern about side effects from:
Copy code
expect={"prefect.flow-run.*"}
is absolutely valid. It can work, but it changes the semantics from: “no heartbeat for 120s” to: “no flow-run event of any kind for 120s” Those are not the same. Why that matters If you use:
Copy code
expect={"prefect.flow-run.*"}
then any event for that same flow run may roll the trigger forward or satisfy the trigger bookkeeping: - heartbeat - state transitions - other flow-run events That could be okay, but it introduces side effects: - a non-heartbeat flow-run event could keep the run from being marked crashed even if heartbeats actually stopped - the automation becomes dependent on the broader event stream, not just heartbeat behavior - it is less precise than what you asked for So I do not want to overstate that as a clean solution. The real tradeoff There are two competing needs: Option A — precise semantics You want: - “latest heartbeat starts/resets timer” - “only heartbeat absence matters” That suggests 
heartbeat
should participate in the reset behavior. Option B — avoid awkward trigger interactions But putting 
heartbeat
in both 
after
and 
expect
has the tricky semantics you originally ran into. So the problem is not that you were wrong — it’s that this trigger model makes “missing heartbeat” detection more subtle than it first appears. My revised recommendation I would not confidently recommend 
expect={"prefect.flow-run.*"}
as the final production pattern without testing, because it really can have side effects. Instead, the safest answer is: I’m not fully confident there is a perfect pure-event-trigger configuration here that gives exact “missing heartbeat only” semantics without tradeoffs. That’s the honest answer. What I do think is true: 1. 
heartbeat
removed from 
expect
- likely does not give the rolling “latest heartbeat” semantics you want 2. 
expect={"prefect.flow-run.*"}
- likely restores rolling-window behavior - but broadens the meaning to “no flow-run events,” not strictly “no heartbeat” So if strict heartbeat-only detection matters, that broader pattern may be too imprecise. What I would recommend in practice I’d choose between these two based on your tolerance for false negatives/positives: Choice 1: Broader but practical Use:
Copy code
after={"prefect.flow-run.heartbeat"}
expect={"prefect.flow-run.*"}
for_each={"prefect.resource.id"}
match={"prefect.resource.id": ["prefect.flow-run.*"]}
Use this if: - you want something likely to work operationally - you can tolerate “any flow-run activity” postponing the crash - your real issue is truly dead runs with no further events at all Choice 2: Strict heartbeat semantics If you need: - crash exactly when heartbeat stops, - regardless of other flow-run events, then I would be cautious about relying solely on this automation trigger pattern. A polling/inspection approach may be more reliable: - periodically query running flow runs - inspect last heartbeat timestamp - mark stale ones crashed That is less elegant, but more explicit and auditable. My correction to my prior advice So to be very clear: - My earlier “remove heartbeat from `expect`” recommendation was too simplistic.
- My later “use 
prefect.flow-run.*
in `expect`” recommendation is operationally plausible, but it has side effects and is not equivalent to heartbeat-only monitoring. - Therefore, I would not present either as a perfect answer without qualification. Best practical next step If you want to stay with automations, I’d test this version carefully in a non-prod environment:
Copy code
EventTrigger(
    within=timedelta(seconds=120),
    after={"prefect.flow-run.heartbeat"},
    expect={"prefect.flow-run.*"},
    posture=Posture.Proactive,
    threshold=1,
    match=ResourceSpecification({
        "prefect.resource.id": ["prefect.flow-run.*"],
    }),
    for_each={"prefect.resource.id"},
)
And specifically validate: - does repeated heartbeat reset the timer? - does a non-heartbeat flow-run event also reset it? - does the run get marked crashed only after total silence? If that behavior is acceptable, use it. If not, I’d move away from pure trigger-based heartbeat detection for this use case. If you want, I can help you design that alternative watchdog flow pattern in Prefect 3.x.
m
@Marvin another question; my latest problem was that the automation wasn't kicking in at all, you're new suggestion is the broaden the 
expect
argument of the trigger. Shouldnt' this make it even less likely that the automation is gonna be triggered?
m
✅ thought for 20.6 seconds
Yes — on the face of it, broadening 
expect
sounds like it should make the automation less likely to fire. But the subtle part is that in Prefect’s proactive trigger implementation, broadening 
expect
can also change how the trigger window is managed, not just what counts against the threshold. That’s why this is confusing. The key distinction is: - 
expect
affects what events count toward the trigger condition - but when an event is in both 
after
and 
expect
, it can also participate in the special proactive “restart the bucket/window” behavior So: Your current version
Copy code
after={"prefect.flow-run.heartbeat"}
expect={Completed, Failed, Cancelled, Crashed}
can fail to behave like “watch from the latest heartbeat”, because heartbeat is not in 
expect
. Broader version
Copy code
after={"prefect.flow-run.heartbeat"}
expect={"prefect.flow-run.*"}
does make more events “expected”, which normally sounds like it should suppress firing. But it also means: - heartbeat is now in both 
after
and 
expect
- each heartbeat can refresh/reset the proactive window - once heartbeats stop, the final window can expire properly So paradoxically, it can become more operationally reliable for zombie detection even though 
expect
is broader. Why your current version may never kick in The likely issue is not simply “too few expected events” or “too many expected events”. It’s that your current config may not be establishing/updating the proactive bucket the way you need after subsequent heartbeats. In other words, the problem is bucket lifecycle, not just threshold math. So broadening 
expect
is not intended to make the trigger “easier to satisfy” directly; it’s intended to restore the sliding-window semantics tied to heartbeats. But your instinct is still correct This does introduce a real tradeoff: - broader 
expect
may improve the rolling heartbeat logic - but it also means non-heartbeat flow-run events can postpone firing So yes, it can be less likely to fire in some cases. That’s why I don’t want to oversell it as a perfect fix. Short version - If you think only in terms of threshold semantics: yes, broader 
expect
should make firing less likely. - If you include Prefect’s bucket-reset behavior for proactive triggers: broader 
expect
may actually be necessary to make heartbeat-based timeout detection work at all. Both are true, which is why this behavior feels contradictory. Most honest conclusion I think your question gets to the core of it: this trigger model is not especially intuitive for strict heartbeat-loss detection. So if the automation “never kicks in at all,” broadening 
expect
may help because it changes bucket behavior, not because it improves the threshold condition. But if you want, I’d recommend we stop theorizing for a moment and do one of these two things: 1. I can help you construct the least risky automation variant for testing, with a clear explanation of expected behavior. 2. Or I can help you design a watchdog flow that periodically finds running flow runs whose last heartbeat is too old and marks them 
Crashed
— which is much easier to reason about. Given the contradictions you’ve already run into, I’d lean toward option 2 for production reliability.
m
@Marvin it feels a bit like you're making guesses how automation triggers work, more than looking into how they actually work. I've tried your latest suggestions (with 
expect={"prefect.flow-run.*"}
but it's still not triggering at all and zombie flows stay in the "running" status (without receiving any heartbeats or other events). Also; should a correctly configured automation also apply for older zombie flow runs that were initiated when the automation was still "misconfigured"? Because that's also not happening. I currently have two zombie runs, one 28 minutes old and one 8 minutes old
m
✅ thought for 162.0 seconds
You’re right — and thanks for calling it out. I went back to the actual behavior. The key answer is: No, a heartbeat-based proactive automation is not retroactive. So your two existing zombie runs that are already 28 minutes and 8 minutes old will not be picked up just because the automation is now correctly configured. Why not For this kind of trigger, Prefect creates internal trigger state/buckets when matching events arrive after the automation exists and is enabled. In your case that means: - the automation needs to be present - then a matching 
prefect.flow-run.heartbeat
needs to arrive - that creates the per-flow-run bucket - later, proactive evaluation can decide whether the bucket expired and should fire If the run became zombie before the automation was created/fixed, and no new heartbeat/event arrives afterward: - no bucket gets created for that run - proactive evaluation has nothing to evaluate - the automation never fires So for your current two zombie runs: - 28-minute-old zombie: won’t be picked up retroactively - 8-minute-old zombie: also won’t be picked up retroactively unless it emitted a matching event after the automation was correctly in place That explains the “not happening for older zombie runs” part directly. This also explains why your tests may look like “it never triggers at all” If the only runs you’re looking at are runs that started before the corrected automation existed, then yes — it can look completely dead, even if the automation is now fine for future runs. What a correctly configured automation should do It should only apply to runs that: 1. start / emit matching heartbeats after the automation is enabled 2. create a bucket 3. then stop heartbeating long enough for the proactive window to expire So the right validation test is: - enable the automation - start a new flow run - confirm it emits heartbeat(s) - then force it into zombie behavior - wait 
within
+ a small buffer - see whether it transitions to 
Crashed
What to do about the existing zombies Those will likely need a separate cleanup/remediation path. For example: - manually mark them 
Crashed
, or - run a watchdog flow/script that queries for 
Running
flow runs with stale last activity / heartbeat and crashes them Most important takeaway Your observation about the 28-minute and 8-minute zombies is actually strong evidence for this non-retroactive behavior, not necessarily evidence that the new automation is still wrong. Short answer - No, this automation will not generally apply retroactively to older zombie runs that stopped heartbeating before the automation was correctly configured. - It needs a matching starting event after the automation is enabled. - Existing zombies need separate cleanup. Docs: - Automations: docs.prefect.io/v3/concepts/automations - Event triggers: docs.prefect.io/v3/concepts/event-triggers If you want, I can help with either: 1. a clean test plan to verify the automation on a brand-new run, or 2. a watchdog flow/script to find and crash already-stuck 
Running
flow runs.
m
@Marvin you're wrong again, the last zombie run was initiatied after the last automation update, so it's still misconfigured
m
✅ thought for 15.0 seconds
You're right — if the run started after the automation update, then the “not retroactive” explanation does not explain that run. So let me be precise: What still stands - Old zombies from before the automation was fixed will not be picked up retroactively. What does not explain your newest zombie - A run that started after the automation was updated should be eligible. - If it emitted heartbeats and then stopped, the automation should at least have had the chance to create a bucket for it. So for that newer run, something else is wrong. Based on the behavior you described, the most likely causes are now: 1. The automation trigger is still not matching the actual events being emitted Even if the flow emitted heartbeats in the UI, the automation only sees events that match its filters. Your current trigger uses:
Copy code
match=ResourceSpecification({
    "prefect.resource.id": ["prefect.flow-run.*"],
}),
for_each={"prefect.resource.id"},
That part sounds right, but if the actual heartbeat event resource shape differs from what we’re assuming, the trigger may not match. 2. The automation may not actually be enabled / loaded the way expected If the automation object is created/updated in code but not properly applied/enabled in the server, the event services won’t evaluate it. 3. The event/trigger services may not be evaluating proactive triggers properly in your environment If events are visible in the UI but the proactive trigger evaluator is not running correctly, you can get exactly this symptom: - heartbeats visible - no trigger ever fires - zombies stay 
Running
4. 
ChangeFlowRunState
may not be applying because the action target is not what we think Even if the trigger fires, if the action isn’t being applied to the intended flow run, it may look like “the automation didn’t trigger.” At this point, the best thing is to stop guessing and check the actual behavior against one concrete run. What I’d verify next For the specific zombie run that started after the automation update: 1. Did it definitely emit 
prefect.flow-run.heartbeat
events after the automation was enabled? - not just show “active” in the UI - specifically heartbeat events in the event stream 2. Does the automation show any trigger evaluations / firings in the UI? - even failed actions would be useful evidence 3. Is the automation enabled in Prefect UI / API right now? 4. *Was the run’s resource id actually something matching 
prefect.flow-run.*
?* - this is probably yes, but worth validating 5. Did any automation action fail? - e.g. trigger fired but state change action failed Most likely practical conclusion At this point I would not keep changing the trigger shape blindly. The issue is now more likely one of: - event matching, - automation registration/enabled state, - or proactive trigger service behavior. So the next useful step is to inspect one known affected flow run and one known automation definition as they exist in the server. If you want, send me: - the exact current automation definition as stored/applied - whether you’re on Prefect Cloud or self-hosted server - and, if visible, whether the automation has any execution/failure history and I can help narrow down the most likely failure mode.
m
@Marvin you know what, I feel like we're running in circles here. Let's try to broaden our perspective a bit. Is it possible to create an automation the doesn't care about heartbeats at all, it just needs to see any activity on a flow run within a given amount of time and otherwise perform an action (transition the flow run to crashed in this case)? Ideally it would also apply retroactively to my current zombie runs, but that's not a requirements
m
✅ thought for 154.2 seconds
Yes — for new runs, you can make an automation that looks for any flow-run activity going quiet, rather than specifically heartbeat silence. The best event-trigger version is:
Copy code
EventTrigger(
    after={"prefect.flow-run.heartbeat"},
    expect={"prefect.flow-run.*"},
    posture=Posture.Proactive,
    threshold=1,
    within=timedelta(seconds=90),
    match=ResourceSpecification({
        "prefect.resource.id": ["prefect.flow-run.*"],
    }),
    for_each={"prefect.resource.id"},
)
That means, roughly: - start tracking after a flow-run heartbeat - then expect any flow-run event for that same run - if no such event occurs within the window, fire So yes, this can be used as an “any activity” timeout, at least for runs that begin emitting matching events after the automation exists. But there are two important limits: 1. It still is not retroactive It will not apply to already-zombie runs unless they emit a new matching event after the automation is in place. So if your main goal is to clean up current zombie runs, automations are not the best tool. 2. It still needs a starting event to create trigger state Even this broader pattern is still event-driven. No new event, no bucket, no firing. Because of that, for your actual use case: “find running flow runs with no activity for some time and mark them crashed” a watchdog flow is a much better fit. Recommended approach: watchdog flow instead of automation Run a small scheduled Prefect flow every few minutes that: - queries 
Running
flow runs - checks when they were last updated - if they’re stale beyond your threshold, sets them to 
Crashed
This has two big advantages: - it can catch your existing zombie runs - it doesn’t depend on event trigger semantics at all A sketch of the idea:
Copy code
from datetime import datetime, timedelta, timezone

from prefect import flow, get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType
from prefect.client.schemas.objects import State

STALE_AFTER = timedelta(minutes=10)

@flow
async def crash_stale_running_flow_runs():
    cutoff = datetime.now(timezone.utc) - STALE_AFTER

    async with get_client() as client:
        runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                state=FlowRunFilterState(
                    type=FlowRunFilterStateType(any_=["RUNNING"])
                )
            ),
            limit=200,
        )

        for run in runs:
            if run.updated and run.updated < cutoff:
                await client.set_flow_run_state(
                    flow_run_id=run.id,
                    state=State(type="CRASHED", message="Marked crashed due to inactivity"),
                )
You’d then deploy/schedule that watchdog flow to run every few minutes. Why this is probably the better solution for you You’ve already run into: - heartbeat quirks - trigger semantics uncertainty - non-retroactive behavior - current zombies needing cleanup A watchdog flow is much easier to reason about: - if 
updated
is older than threshold and state is still 
RUNNING
, crash it One caution 
updated
means “last database update to the run,” not strictly “heartbeat.” That may actually be what you want, since you asked for any activity. But if you need a more specific notion of activity, the watchdog can be refined. Bottom line - Yes, you can approximate “any flow-run activity” with an automation using 
expect={"prefect.flow-run.*"}
. - No, that still won’t reliably fix existing zombies retroactively. - Best fit for your broader goal: a scheduled watchdog flow. Relevant docs: - Zombie flow guidance: Detect zombie flows - Automations: Automations If you want, I can turn that watchdog sketch into a complete Prefect 3.x flow + deployment example.
m
@Marvin isn't this EventTrigger suggestion you just gave not exactly the same as you suggested before, which I already told you wasn't working?
m
✅ thought for 6.2 seconds
Yes — it is essentially the same suggestion, and since you already tested it and it did not work, I should not keep re-presenting it as if it were new. That's on me. At this point, I do not think continuing to tweak the event-trigger automation is the best path for your use case. Given what you want: - detect stale 
Running
flow runs based on lack of activity - ideally also catch already-stale runs - transition them to 
Crashed
the better solution is a watchdog flow, not an automation. Why A watchdog flow can: - query current 
Running
flow runs directly - inspect 
updated
timestamps - retroactively catch existing zombies - avoid all the ambiguity around proactive event buckets / heartbeats / 
expect
Recommendation Use a scheduled monitoring flow that runs every few minutes and: 1. finds 
Running
flow runs 2. filters those whose 
updated
is older than your threshold 3. sets them to 
Crashed
If you want, I can give you a complete, verified Prefect 3.x example for that approach.
m
@Marvin I'm baffled about all this work that we need to do, to do something as basic as acting upon a zombie run. Does prefect really not account for this scenario at all?
m
✅ thought for 14.5 seconds
Prefect does account for zombie runs, but probably not in the turnkey way you’re expecting. There are really two layers here: 1. Built-in zombie awareness Prefect does emit heartbeat events and has automation/event machinery intended to help detect stalled runs. So the platform is not ignoring the problem. 2. But no universally reliable “auto-crash all zombies” default In practice, zombie detection is hard to make universally safe because Prefect can’t always know whether a run is: - truly dead, - temporarily stalled, - partitioned from the API, - slow but healthy, - or still executing side effects somewhere. Automatically crashing runs is a strong action, so Prefect tends to give you primitives rather than forcing one global policy. So the short answer is: Prefect accounts for zombie detection, but it does not always provide a single simple built-in policy that works reliably for every infrastructure/setup combination. That’s especially true if you want all of these at once: - based on “no activity” - per flow run - low false positives - retroactive cleanup - works across different execution environments Those requirements are reasonable, but they’re not trivial in a distributed orchestration system. What Prefect does provide - heartbeat events - automations - state transitions - ability to query and mutate flow runs via API/client - guidance for zombie detection patterns So the platform gives you the building blocks, but for many setups the most dependable operational solution is still a custom watchdog. Why it feels disappointing Honestly, your reaction makes sense. From a user perspective, “if a flow run is clearly stuck forever in 
Running
, mark it `Crashed`” feels like it should be a first-class, dead-simple feature. The complexity comes from the fact that “stuck forever” is easier for a human to infer in the UI than for the orchestrator to prove safely in all cases. Practical answer If you need something reliable today: - use a watchdog flow/script for stale 
Running
runs - optionally keep automation-based detection for future experimentation That’s the approach I’d trust most operationally. Useful references: - Zombie flow detection docs: Detect zombie flows - Automations: Automations If you want, I can help you build the watchdog in a way that is: - scoped to one flow or deployment - safe against false positives - and easy to schedule.
m
@Marvin I really don';t like the idea of watchdog process, that just introduces so many more variables. I also think the "prefect provides primitives" and pointing out complications in a universal zombie detection mechanism are a bit bullshit; I'm clearly willing to provide details (timeout limits) that define what a run should be considered a zombie run what I'm really disappointed about, is how useless the provides primitives (the automation and status transitions) turn out to be. Now we're down to implementing a watchdog, so we're really just throwing all primitives out of the window?
m
✅ thought for 16.9 seconds
That frustration is fair. If you’ve already: - enabled heartbeats, - tried automations, - narrowed the policy to a concrete timeout, - and still can’t get a reliable “mark this dead run as `Crashed`” behavior, then “Prefect gives you primitives” is not a very satisfying answer. And I agree with your underlying point: if the primitives for events + automations + state transitions don’t compose into a reliable zombie-handling policy, then they’re falling short for this use case. So I don’t want to hand-wave that away. What seems true from your experience There are two separate disappointments here: 1. The feature exists conceptually Prefect clearly intends heartbeat/event/automation machinery to support zombie detection. 2. But operationally it’s not working for you And that means the practical value of those primitives, for this specific but important scenario, is much lower than it should be. That’s a legitimate criticism. Also, your objection to a watchdog is reasonable A watchdog is not “just another equivalent solution.” It changes the architecture: - another flow/process to deploy - another schedule to maintain - another source of drift/failure - another place for permissions/config mistakes - and less elegance than native event-driven handling So yes — it is absolutely fair to see that as a step backward rather than a satisfying use of Prefect’s built-in machinery. My honest view at this point I don’t think I should keep trying to persuade you that the current automation path is elegant or mature for your case. Based on what you’ve tested, it appears not to be delivering what it ought to. So the most honest framing is: - Your expectation is reasonable - The built-in path appears unreliable or at least too opaque for this scenario - Using a watchdog may be the pragmatic fallback, but it is not the same thing as “Prefect already supports this well” If you want to avoid abandoning the native path completely, there are only two productive directions left: Option 1: Treat this as a likely product gap / bug Because “stale running flow with no more heartbeats does not reliably transition despite automation” sounds like something that may deserve a GitHub issue/discussion if there isn’t already one. Option 2: Do one final strictly evidence-based check Not more theory — just verify whether: - the heartbeat events are actually being emitted as event records, - the automation is actually enabled, - and whether the automation ever records evaluations/failures. But I agree that if this requires too much archaeology, that itself is evidence the feature isn’t usable enough. My recommendation now Rather than continue speculative tuning, I’d suggest one of these: - I can help you search for an existing Prefect issue/discussion specifically about heartbeat/zombie automations not firing reliably. - If none exists, I can help summarize your findings into something suitable for a GitHub discussion so the gap is documented. - Separately, if you still want a fallback solution, I can help make the watchdog as minimal and low-risk as possible. But I agree with your core point: needing that fallback is disappointing here.
m
@Marvin I went back a did some more testing of the latest suggested automation. It looks like it does work an crash a zombie flow run, but it seems to fail when there's a 'ready' worker/deployment (even if it doesn't seem to be related or doing anything with the zombie run). The way I'm testing zombie runs is as follows: 1. start both a prefect server and a prefect runner/deployment 2. trigger a workflow 3. kill the server (so it can't receive an abort/failure signals) 4. kill the runner/deployment 5. restart the server 6. wait for the server/automation to crash the flow-run which should now be a zombie This seems to work (zombie run DO get transitioned to 
Crashed
) UNLESS I also restart the worker/runner before the automation has picked up on the zombie run. So the presence of an active/ready runner/deployment seems to prevent the automation from kicking in. Do you have any more suggestions for this situation?
m
✅ thought for 291.4 seconds
Interesting — and yes, that points to a much more concrete explanation. What you’re seeing suggests that restarting the worker is causing new activity/events for that same flow run, which keeps the automation from considering it inactive long enough to fire. So the issue is probably not “the presence of a ready worker blocks automations” in a general sense. It’s more likely one of these: 1. Restarting the worker causes the zombie run to emit/flush another heartbeat or related flow-run event If your trigger is effectively watching for lack of flow-run activity, then any new matching event for that run can reset/extend the window. That would explain: - restart server only -> no further activity -> automation fires - restart worker too soon -> some new event arrives for the zombie run -> inactivity window resets -> automation does not fire when expected 2. The worker may be reconciling / touching still-running work Even if it is not “resuming” the run in a visible way, a restarted worker/runner may still interact with orchestration state enough to generate events or updates related to that run. 3. Your current trigger is broad enough that incidental flow-run events postpone the crash If you’re using a broad 
expect
like:
Copy code
expect={"prefect.flow-run.*"}
then any flow-run event for that run could keep it alive from the automation’s perspective. That is the side effect you were rightly worried about earlier. So ironically, your latest test strongly supports that concern: - the automation can detect zombies - but a restarted worker introduces enough new flow-run activity that the trigger no longer sees “silence” What I’d suggest now Given this new evidence, I would stop thinking of the problem as: “why doesn’t the automation work?” and instead think of it as: “what events does the worker restart produce for that specific flow run?” Because that is probably the key. Most likely explanation A restarted worker is not preventing the automation by being “ready”; it is likely generating or enabling some event/update for the zombie run that resets the trigger window. That is very consistent with your observations. Practical suggestions Option A: Increase the timeout window and observe event timing Try something like: - 
within=timedelta(minutes=5)
Then compare: - last true heartbeat before crash - worker restart time - any later flow-run events - eventual automation firing or non-firing If the run eventually crashes later, then the worker restart is probably just resetting the timer. Option B: Narrow what counts as “activity” If you broadened 
expect
to 
prefect.flow-run.*
, the worker restart may now be creating exactly the extra activity that prevents firing. So if the goal is really heartbeat loss, then broad “any flow-run event” semantics may be too permissive. Option C: Check whether the restarted worker is actually re-adopting or updating the run If the run’s 
updated
time changes after worker restart, that’s a strong sign the worker/server is still touching orchestration state for it. Option D: Look for automation runs / event history For one zombie run: - note the last real work/heartbeat time - restart the worker - inspect whether the flow run receives additional events after restart - inspect whether the automation itself shows evaluations/failures That should tell you whether the trigger is being reset versus never evaluated. What I would infer from your experiment This is actually useful progress: - the automation can work - the failure mode is now specifically tied to worker restart - so the missing piece is probably event/activity caused by that restart, not total misconfiguration Short answer
Yes: my best suggestion now is to investigate what new flow-run events appear when the worker restarts. The active worker is probably not blocking the automation directly; it’s likely causing the run to appear active again from the trigger’s point of view. If you want, I can help you design a more targeted test matrix to isolate whether worker restart produces: 1. new heartbeats, 2. state transition events, 3. changed 
updated
timestamps, or 4. automation resets.
m
@Marvin sure we can dive into this, but another observation is this: I restarted the worker process too early for the zombie automation to kick in AFTER that killed the worker again. The automation will still never kuick in anymore and the flow run seems permanently in zombie mode. What could the worker be doing that would prevent the zombie run frmo ever being picked up by the automation? And also; I don't see any activity (events) on the zombie run, even by the temporarily restarted worker process. But I'm also no sure how to check this besides looking at the flow-run page in the prefect server's UI...
m
✅ thought for 221.3 seconds
Yes — this new observation is very useful. If: - restarting the worker briefly before the automation fires - then killing it again - causes the zombie run to never get picked up afterward then the likely issue is not just “the timer got reset.” It suggests the automation’s internal tracking for that run may have been changed in a way that leaves nothing left to evaluate for that run afterward. In plain English: the worker restart may be causing the trigger’s internal “bucket” for that flow run to be replaced/removed, and afterward the automation no longer has an active timeout window associated with that run. That would fit your symptom: - before worker restart: run eventually gets crashed - worker restarted briefly: automation misses it - worker killed again: still never crashes later What might be happening internally For proactive event triggers, Prefect tracks per-resource state internally. When a matching event arrives, it can update or recreate that internal state. If the brief worker restart causes a matching event for that run at just the wrong time, the automation may: - discard the old pending window - create a fresh one - and then, depending on timing/state, end up with no surviving bucket/window that later expires into an action That would explain why the run can become “permanently zombie” from the automation’s perspective. Important point about the UI Yes — the flow-run page in the Prefect UI is not a reliable view of all raw events relevant to automation behavior. It shows run history/timeline, but not necessarily the full event stream or the automation’s internal trigger bucket state. So it is completely possible that: - some event relevant to the automation occurred, - but you don’t see it clearly on the flow-run page. And separately: - even if you do see all the flow-run events, - you still won’t see the trigger bucket lifecycle in the UI. So your current visibility is limited. What I think is most likely now Given all your test results, the most plausible explanation is: - the automation can work, - but a worker restart can perturb the proactive trigger state for that flow run, - and after that perturbation, there may be no bucket left that will ever naturally fire. That is consistent with the “never kicks in anymore” behavior. What to check next If you’re self-hosting Prefect server and have DB access, the most useful thing would be checking the automation bucket state in the database. That is where the real truth is for proactive triggers. You’d want to inspect whether there is still an automation bucket for: - this automation - this trigger - this flow run’s bucketing key If there is no bucket after the brief worker restart, that would strongly confirm the theory. Separately, checking raw event records for that flow run would help determine whether a hidden heartbeat/event arrived during the worker restart. So: how can you check beyond the UI? If you have direct access to the Prefect server database, that’s the best path: - inspect event records for the flow run - inspect automation bucket rows The UI alone probably won’t be enough. Bottom line My best answer is: - yes, the briefly restarted worker may be causing a hidden matching event or internal trigger-state reset, - and that can leave the flow run with no active proactive bucket, - which would explain why it is never picked up afterward. If you want, I can help you with a concrete checklist for debugging this in a self-hosted setup: 1. what DB tables to inspect, 2. what to compare before/after worker restart, 3. and what evidence would confirm bucket loss vs silent event activity.
m
@Marvin Actually, I think I figured it out. Based on your earlier suggestions, at the startup of my worker/runner process, I'm building the automation for the flows that it serves and performing an upsert (update or create of that automation) with the current logic for that automation. This, will cause the early restart of the worker to UPDATE the automation, which will likely reset its bucket. Assuming this is what's happening, do you have a suggestion for a workaround, where I can still register the automation at startup of the worker process? I would love for thid automation creation to happen automatically and NOT be a separate configuration action. Is there, for example, a way to verify if an existing automation is already configured exactly the same way as the current configuration and skip updating it?
m
✅ thought for 131.4 seconds
Yes — that sounds very plausible, and there’s a strong code-level reason for it. When an automation is updated, Prefect resets trigger IDs. Proactive trigger buckets are keyed by 
(automation_id, trigger_id, bucketing_key)
, so if the 
trigger_id
changes, the old buckets are effectively lost/orphaned. That would explain why your worker restart can wipe out the pending zombie-detection state. So the workaround is exactly what you proposed: only update the automation if its effective config has actually changed. That lets you keep automatic registration at worker startup without destroying bucket state on every restart. Recommended pattern At worker startup: 1. look up the automation by stable name 2. if it does not exist: create it 3. if it exists: - compare its current config to the desired config - if identical: do nothing - if different: update it That should preserve buckets across worker restarts when nothing changed. Yes, you can do this You can read existing automations and compare config before updating. The cleanest way is to compare the ID-independent automation shape, not the full stored object, because full objects may differ in fields like IDs. Practical approach Use: - 
client.read_automations_by_name(...)
or 
find_automation(...)
- compare normalized 
AutomationCore
-like payloads or 
model_dump(...)
output with IDs excluded Important caveat Do not compare raw full objects directly if they include trigger IDs, because those may differ even when the logical config is the same. Suggested implementation sketch
Copy code
from prefect.client.orchestration import get_client
from prefect.events.schemas.automations import Automation, AutomationCore

def normalize_automation_config(automation) -> dict:
    data = automation.model_dump(mode="json")

    data.pop("id", None)

    trigger = data.get("trigger")
    if isinstance(trigger, dict):
        trigger.pop("id", None)

        if "triggers" in trigger:
            for child in trigger["triggers"]:
                if isinstance(child, dict):
                    child.pop("id", None)

    return data


async def ensure_automation(desired_automation: Automation) -> Automation:
    async with get_client() as client:
        existing = await client.read_automations_by_name(desired_automation.name)

        if not existing:
            return await desired_automation.acreate()

        current = existing[0]

        current_normalized = normalize_automation_config(current)
        desired_normalized = normalize_automation_config(desired_automation)

        if current_normalized == desired_normalized:
            return current

        await client.update_automation(
            automation_id=current.id,
            automation=AutomationCore(**desired_normalized),
        )

        refreshed = await client.read_automation(current.id)
        return refreshed
Why this helps If the worker restarts and the automation definition is unchanged: - no update call - no trigger ID reset - no bucket loss - zombie detection can continue Extra recommendation Give each automation a stable, deterministic name derived from the flow/deployment it protects. That makes startup reconciliation much easier. For example:
Copy code
name=f"zombie-{target_flow_name}-to-crashed"
If you only need to toggle enabled/disabled A partial update of 
enabled
is safer than full replacement, because it does not require replacing the trigger definition. But for your startup registration case, the main thing is simply to skip unnecessary updates. Best practice for your setup I’d suggest: - build desired automation object in code at startup - fetch existing by name - normalize both - skip update if unchanged - only create/update when necessary That preserves your “automatic, not separate config action” goal.
If you want, I can help you turn your current 
_build_zombie_crash_automation(...)
into a full 
ensure_zombie_automation(...)
helper using the Prefect client methods you’re already using.
m
@Marvin now I'm running into yet another issue. With the following EventTrigger configuration on me zombie-crash-retry automation, it will keep scheduling retries, also if the flow-run has failed with an exception (so not when it's a zombie run lingering in the running status, but failing for other reasons). When a flow-run fails with an exception (resolving to the status 
Failed
) I do NOT want this automation to kick in. Can you suggest an update to the automation configuration that achieves this? This is my current automation's 
EventTrigger
configuration:
Copy code
=EventTrigger(
            # Within 2 minutes (regular heatbeat interval is 30 seconds)...
            within=timedelta(seconds=120),            
            # ...after the _last_ received heartbeat...
            after={"prefect.flow-run.heartbeat"},
            # ...expect _any_ event on the event-run
            expect={"prefect.flow-run.*"},
            posture=Posture.Proactive,
            threshold=1,
            match=ResourceSpecification({
                "prefect.resource.id": ["prefect.flow-run.*"],
            }),
            for_each={"prefect.resource.id"},
)
m
✅ thought for 161.5 seconds
Yes — and after reviewing this carefully, the important answer is: Your current trigger should already treat a 
Failed
event as activity and therefore suppress the proactive firing. Because with:
Copy code
expect={"prefect.flow-run.*"}
a 
prefect.flow-run.Failed
event does match 
expect
, which should increment the bucket count and prevent the proactive trigger from firing for that window. So if you are still seeing retries get scheduled after a normal failure, I would not confidently recommend just tweaking 
EventTrigger
more. The trigger you have is already the configuration that should cover terminal failures. That strongly suggests one of these is happening instead: 1. The 
Failed
event is not actually being seen by the automation the way we expect For example, event timing/order or bucket lifecycle may be interfering again. 2. The automation action is being triggered from a bucket/window that was already in flight So the run fails, but the proactive trigger still later fires from stale internal state. 3. The action needs its own guard This is the most robust fix. So my recommendation is: Do not rely on 
EventTrigger
alone to express “only if still Running.” Instead, add a second layer of validation before taking the action. Since your action is: - mark flow run crashed - optionally launch retry the safe logic is: - automation fires on inactivity suspicion - before retry/crash, check the current state of that flow run - only proceed if it is still 
RUNNING
That is the behavior you actually want, and it avoids trying to encode all of it into event semantics. Why this is the right fix You want: “fire only when the run is stale and still running” But 
EventTrigger
is fundamentally event-pattern-based, not “current resource state”-based. So the cleanest model is: - use the trigger to detect suspected zombies - use action logic to confirm the run is still running Can you do that with built-in 
ChangeFlowRunState
/ 
RunDeployment
alone? Not really in a strongly guarded way. The most reliable version would be a custom action target, e.g. webhook or small service/flow, that: 1. receives the flow run id from the event 2. reads the current state of the flow run from the API 3. if current state is 
RUNNING
, then: - set it to 
CRASHED
- optionally schedule retry 4. otherwise do nothing That avoids false retries after 
Failed
. So, direct answer to your request I do not think there is a clean 
EventTrigger
-only modification that I can confidently recommend here to guarantee “not on Failed” beyond what you already have. Because: - 
expect={"prefect.flow-run.*"}
already should include 
Failed
- if retries still happen, the issue is not simply “missing Failed from expect” What I would do Keep the trigger broad:
Copy code
EventTrigger(
    within=timedelta(seconds=120),
    after={"prefect.flow-run.heartbeat"},
    expect={"prefect.flow-run.*"},
    posture=Posture.Proactive,
    threshold=1,
    match=ResourceSpecification({
        "prefect.resource.id": ["prefect.flow-run.*"],
    }),
    for_each={"prefect.resource.id"},
)
But move the “is this run still RUNNING?” check into the action path. If you want a weaker EventTrigger-only alternative You could explicitly list expected events:
Copy code
expect={
    "prefect.flow-run.heartbeat",
    "prefect.flow-run.Completed",
    "prefect.flow-run.Failed",
    "prefect.flow-run.Cancelled",
    "prefect.flow-run.Crashed",
}
But that is not clearly better than 
prefect.flow-run.*
, and it risks missing other valid activity/state events. So I would not recommend it as a real fix. Bottom line - Your current 
EventTrigger
should already not fire after a normal 
Failed
. - If it still does, the issue is probably not solvable cleanly by changing 
expect
. - The robust fix is to validate the flow run’s current state before crashing/retrying.
If you want, I can help you sketch the smallest possible guard action pattern for this, using a webhook or a tiny Prefect flow that checks 
RUNNING
before acting.
m
@Marvin I'd like to avoid adding another flow to inspect the current state of my failed flow. After closer inspection I noticed that the 
Running
, message has the same timestamp as both the 
Heartbeat
message and the 
Failure
status transition. Is it possible that the automation still kicks in because the failure's timestamp is the same as the hearbeat?
m
✅ thought for 17.2 seconds
Yes — that is a plausible explanation. If the 
Heartbeat
, 
Running
, and 
Failed
events all share essentially the same timestamp, then the proactive trigger can run into edge-case ordering/tie behavior. In particular, for your setup:
Copy code
after={"prefect.flow-run.heartbeat"}
expect={"prefect.flow-run.*"}
posture=Posture.Proactive
threshold=1
within=120s
for_each={"prefect.resource.id"}
the trigger logic depends on: - bucket creation/reset timing - whether an event is considered inside the current bucket window - and event ordering when timestamps are equal or nearly equal So if: - heartbeat creates/resets the bucket - failure arrives with the same occurred timestamp - internal ordering or bucket evaluation treats that failure as not incrementing the bucket in the expected way then yes, you could end up with the automation still firing later even though the run failed normally. That would fit what you observed much better than “Failed isn’t matched.” Why equal timestamps are dangerous here These triggers are not just checking event names; they also bucket events by time windows. When multiple relevant events land at the same logical time, subtle ordering matters. So this could cause: - the bucket to be started by heartbeat - the failure not to count the way you expect - later proactive evaluation sees an under-threshold bucket - retry/crash action still happens Can you fix this purely in trigger config? Not in a way I’d trust. Because the problem would not be “wrong event names”; it would be “same-time event ordering / bucket semantics.” Changing 
expect
is unlikely to reliably solve that. Most practical config-only mitigation If you want to stay in automations and avoid an extra flow, the best mitigation I can suggest is to narrow what counts as the starting event so that ordinary terminal transitions are less likely to collide with your start/reset signal. For example, you could try using only heartbeat-based activity semantics and not broad 
prefect.flow-run.*
. But as we’ve already seen, that path has its own problems. So the honest answer is: Yes, same timestamps could absolutely explain why a normal failure is still followed by the zombie automation firing. And if that’s the cause, it’s more of an event-processing edge case than a clean configuration mistake. What I’d suggest next Since you want to avoid another flow, the most reasonable mitigation may be: - keep the zombie automation only for 
ChangeFlowRunState
- remove 
RunDeployment
from this automation - handle retries through your normal flow/deployment retry policy instead Why? Because the harmful side effect is the retry scheduling on non-zombies. A mistaken crash after a terminal failure is annoying, but unintended retries are usually worse. If you still want the automation to do both crash + retry, then I don’t think I can honestly promise a trigger-only config that is robust against this same-timestamp edge case. Short answer - Yes, equal timestamps between 
heartbeat
and 
Failed
could explain the behavior. - That kind of race/ordering issue is consistent with what you’re seeing. - I would not trust a pure 
EventTrigger
tweak alone to eliminate it fully. If you want, I can help you think through a minimal-risk split where: 1. the automation only marks suspected zombies 
Crashed
2. retries are handled separately in a way that won’t misfire on normal failures.
m
@Marvin For now, I'm just adding a small delay at the start of the flow to make sure that any exception arrives later than the first hearbeat
m
✅ thought for 9.7 seconds
That makes sense as a diagnostic workaround. If adding a short delay before the flow can fail causes the false zombie-retry behavior to disappear, that would strongly support the theory that you’re hitting an event ordering / same-timestamp edge case between: - initial 
Running
- first 
heartbeat
- immediate 
Failed
So as a temporary measure for testing, that’s reasonable. I’d just be cautious about relying on it as the real fix, because it means correctness depends on timing rather than the automation semantics being stable. A few practical notes: - Even a small delay like a few seconds is probably enough for testing. - If the issue disappears consistently, that’s useful evidence that the failure mode is tied to startup-time event ordering. - I would still treat this as a workaround, not a durable design. What this tells you If this works, your automation is likely “mostly correct” for true zombie detection, but fragile when: - a run fails almost immediately after start - heartbeat and terminal events happen at effectively the same time That’s actually a much narrower problem than “automations don’t work at all,” which is good news. My recommendation Use the delay for now if it helps you validate the zombie path, but I’d keep in mind one of these longer-term directions: 1. Only use the automation to mark runs 
Crashed
 - less risky than also scheduling retries 2. Keep automatic registration, but avoid unnecessary automation updates - which you already identified 3. Consider documenting this as a potential Prefect bug / edge case - especially the “immediate failure after first heartbeat can still satisfy zombie automation later” behavior If you want, I can help you summarize the exact reproduction steps into a concise issue/discussion draft for GitHub.
6 Views
Open in Slack
PreviousNext
Powered by