Kestra stuck paused execution recovery guard

id: kestra-stuck-paused-execution-recovery-guard
namespace: company.team

description: |
  Find Kestra executions that have been sitting PAUSED past a grace window, the approvals nobody
  ever answered. List them in dry run. On auto_remediate, behind an allowlist, a ceiling and a Pause
  approval of its own, answer each abandoned approval with the Kestra plugin, then re-query and fail
  if any targeted execution is still PAUSED.

inputs:
  - id: kestra_url
    type: STRING
    displayName: Target Kestra API URL
    description: The Kestra whose paused executions are recovered. The default
      points the flow at its own cluster.
    defaults: "http://localhost:8080"

  - id: namespace
    type: STRING
    displayName: Namespace to scan
    description: Only executions in exactly this namespace are considered.
    defaults: company.team

  - id: min_paused_minutes
    type: INT
    displayName: Grace window (minutes)
    description: An execution must have been PAUSED longer than this before it
      counts as abandoned. A pause younger than the window is reported as still
      waiting for its human.
    defaults: 60

  - id: allow_flows
    type: ARRAY
    itemType: STRING
    displayName: Flows safe to resume
    description: Flow ids whose pauses you have decided are safe to answer. An empty
      list means nothing is resumed, which is the safe default. This is the
      control that keeps an accidental abandon apart from a deliberate refusal.
    defaults: []

  - id: required_pause_task
    type: STRING
    displayName: Expected paused task id
    description: When set, an execution is only answered if the task it is paused on
      carries this id. A candidate paused anywhere else is left alone, so the
      re-verify fails the run and a human looks at it. Leave empty to skip the
      task check.
    defaults: ""

  - id: resume_answer
    type: STRING
    displayName: Answer sent to the pause
    description: JSON object posted as the onResume answer, so the answer is
      something you wrote rather than something this flow invented. Must match
      the onResume inputs of the flows in allow_flows.
    defaults: '{"approved": true}'

  - id: max_resumes
    type: INT
    displayName: Resume ceiling
    description: Refuse to answer more than this many pauses in one run. A bigger
      pile is an incident for a human, not a batch job.
    defaults: 5

  - id: kestra_user
    type: STRING
    displayName: Kestra API username
    defaults: admin@kestra.io

  - id: kestra_password
    type: STRING
    displayName: Kestra API password
    defaults: "{{ secret('KESTRA_PASSWORD') }}"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow the resume. Still requires dry_run false, a non-empty
      allowlist, a count under the ceiling plus a Pause approval.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any pause to be answered.
    defaults: true

variables:
  map_program: '[.[][] | {id, flowId, paused: ([.state.histories[] | select(.state
    == "PAUSED") | .date] | last)}]'
  stale_program: "{{ vars.map_program ~ ' | map(select(.flowId != \"' ~ flow.id ~
    '\"))' ~ ' | map(select(.id != \"' ~ execution.id ~ '\"))' ~ ' |
    map(select(.paused <= \"' ~ outputs.window.values.cutoff ~ '\"))' }}"
  allow_program: "{{ 'map(select(([.flowId] - ' ~ (inputs.allow_flows | jq('.') |
    first | toJson) ~ ') | length == 0))' }}"
  paused_task_program: '[.taskRunList[] | select(.state.current == "PAUSED") | .taskId] | last'

tasks:
  - id: window
    type: io.kestra.plugin.core.output.OutputValues
    description: Freeze the grace window once, so every later step compares against
      the same instant.
    values:
      cutoff: "{{ now(timeZone='UTC') | dateAdd(-1 * inputs.min_paused_minutes,
        'MINUTES') | date(\"yyyy-MM-dd'T'HH:mm:ss.SSS'Z'\", timeZone='UTC') }}"

  - id: detect
    type: io.kestra.plugin.kestra.executions.Query
    description: Read-only. Fetch every PAUSED execution in the namespace from the
      target Kestra, with its state history so the time it went PAUSED is known.
    kestraUrl: "{{ inputs.kestra_url }}"
    auth:
      username: "{{ inputs.kestra_user }}"
      password: "{{ inputs.kestra_password }}"
    namespace: "{{ inputs.namespace }}"
    states:
      - PAUSED
    fetchType: FETCH
    size: 200

  - id: paused_total
    type: io.kestra.plugin.kestra.executions.Count
    description: Read-only second opinion on the same cluster, so the detail query
      is checked against the target's own count.
    kestraUrl: "{{ inputs.kestra_url }}"
    auth:
      username: "{{ inputs.kestra_user }}"
      password: "{{ inputs.kestra_password }}"
    namespaces:
      - "{{ inputs.namespace }}"
    states:
      - PAUSED

  - id: select
    type: io.kestra.plugin.core.output.OutputValues
    description: Map every paused execution to the moment it went PAUSED, drop this
      flow and this execution, then keep the ones past the grace window and
      narrow those to the allowlist.
    values:
      all_paused: "{{ outputs.detect.rows | toJson | jq(render(vars.map_program)) |
        first | toJson }}"
      stale: "{{ outputs.detect.rows | toJson | jq(render(vars.stale_program)) | first
        | toJson }}"

  - id: pick
    type: io.kestra.plugin.core.output.OutputValues
    description: Apply the allowlist to the stale set.
    values:
      eligible: "{{ outputs.select.values.stale | jq(render(vars.allow_program)) |
        first | toJson }}"

  - id: counts
    type: io.kestra.plugin.core.output.OutputValues
    values:
      all_count: "{{ outputs.select.values.all_paused | jq('length') | first }}"
      stale_count: "{{ outputs.select.values.stale | jq('length') | first }}"
      eligible_count: "{{ outputs.pick.values.eligible | jq('length') | first }}"
      eligible_ids: "{{ outputs.pick.values.eligible | jq('[.[].id]') | first | toJson }}"
      stale_detail: "{{ outputs.select.values.stale | jq('map(.flowId + \" \" + .id +
        \" paused since \" + .paused) | join(\"; \")') | first }}"
      eligible_detail: "{{ outputs.pick.values.eligible | jq('map(.flowId + \" \" +
        .id) | join(\"; \")') | first }}"
      held_detail: "{{ outputs.select.values.stale | jq(render(vars.allow_program) ~ '
        as $ok | map(select(([.id] - [$ok[].id]) | length > 0)) | map(.flowId +
        \" \" + .id) | join(\"; \")') | first }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Paused guard on {{ inputs.kestra_url }} namespace {{ inputs.namespace }}:
      target counts {{ outputs.paused_total.total ?? outputs.paused_total.count
      ?? 'n/a' }} PAUSED, query returned {{ outputs.counts.values.all_count }},
      {{ outputs.counts.values.stale_count }} past the {{
      inputs.min_paused_minutes }} minute window (cutoff {{
      outputs.window.values.cutoff }}), {{ outputs.counts.values.eligible_count
      }} on the allowlist. Stale: {{ outputs.counts.values.stale_detail }}. Held
      back by the allowlist: {{ outputs.counts.values.held_detail }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.counts.values.stale_count | number) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kestra paused guard on ' ~ inputs.kestra_url ~ ' namespace ' ~ inputs.namespace ~ ': ' ~ outputs.counts.values.stale_count ~ ' execution(s) PAUSED over ' ~ inputs.min_paused_minutes ~ ' min (' ~ outputs.counts.values.stale_detail ~ '). On the allowlist: ' ~ outputs.counts.values.eligible_count ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Answering them behind approval.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: eligible_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ (outputs.counts.values.eligible_count | number) > 0 }}"
            then:
              - id: ceiling
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.counts.values.eligible_count | number) >
                  inputs.max_resumes }}"
                then:
                  - id: too_many
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "Refusing to answer {{ outputs.counts.values.eligible_count }}
                      abandoned pauses, over the max_resumes ceiling of {{
                      inputs.max_resumes }}. That many at once is a process
                      problem, so look at why the approvals are not being
                      answered before raising the ceiling."
                else:
                  - id: approval
                    type: io.kestra.plugin.core.flow.Pause
                    description: Hold for a human to approve answering somebody else's approval. No
                      answer within the window changes nothing.
                    pauseDuration: PT1H
                    behavior: RESUME
                    onResume:
                      - id: approved
                        type: BOOL
                        required: false
                        description: Set true to send the answer to the listed pauses. Anything else
                          leaves them paused.
                      - id: note
                        type: STRING
                        required: false
                        description: Optional reason, kept in the run log.

                  - id: approved_gate
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ outputs.approval.onResume.approved ?? false }}"
                    then:
                      - id: recover
                        type: io.kestra.plugin.core.flow.Loop
                        description: One pause at a time. Remediation is serial so two answers cannot
                          race on the same cluster.
                        concurrencyLimit: 1
                        values: "{{ outputs.counts.values.eligible_ids }}"
                        tasks:
                          - id: detail
                            type: io.kestra.plugin.core.http.Request
                            description: Read-only. Fetch the execution so the task it is paused on is known
                              before anything is answered.
                            uri: "{{ inputs.kestra_url }}/api/v1/main/executions/{{ item.value }}"
                            method: GET
                            contentType: application/json
                            options:
                              auth:
                                type: BASIC
                                username: "{{ inputs.kestra_user }}"
                                password: "{{ inputs.kestra_password }}"

                          - id: task_gate
                            type: io.kestra.plugin.core.flow.If
                            condition: "{{ inputs.required_pause_task == '' or (outputs.detail.body |
                              jq(render(vars.paused_task_program)) | first) ==
                              inputs.required_pause_task }}"
                            then:
                              - id: answer
                                type: io.kestra.plugin.kestra.executions.Resume
                                description: Send the operator's answer to the abandoned approval.
                                allowFailure: true
                                kestraUrl: "{{ inputs.kestra_url }}"
                                auth:
                                  username: "{{ inputs.kestra_user }}"
                                  password: "{{ inputs.kestra_password }}"
                                executionId: "{{ item.value }}"
                                inputs: "{{ inputs.resume_answer }}"
                            else:
                              - id: wrong_task
                                type: io.kestra.plugin.core.log.Log
                                level: WARN
                                message: "Leaving {{ item.value }} paused: it is waiting on {{
                                  outputs.detail.body |
                                  jq(render(vars.paused_task_program)) | first
                                  }}, not on {{ inputs.required_pause_task }}."

                      - id: settle
                        type: io.kestra.plugin.core.flow.Sleep
                        duration: PT5S

                      - id: reverify
                        type: io.kestra.plugin.kestra.executions.Query
                        description: Read the target again. This is what decides whether the answer
                          landed, because the resume task goes green either way.
                        kestraUrl: "{{ inputs.kestra_url }}"
                        auth:
                          username: "{{ inputs.kestra_user }}"
                          password: "{{ inputs.kestra_password }}"
                        namespace: "{{ inputs.namespace }}"
                        states:
                          - PAUSED
                        fetchType: FETCH
                        size: 200

                      - id: verify_counts
                        type: io.kestra.plugin.core.output.OutputValues
                        values:
                          still_ids: "{{ outputs.reverify.rows | toJson | jq('[.[][].id] as $p | ' ~
                            outputs.counts.values.eligible_ids ~ ' - (' ~
                            outputs.counts.values.eligible_ids ~ ' - $p)') |
                            first | toJson }}"
                          still_count: "{{ outputs.reverify.rows | toJson | jq('[.[][].id] as $p | (' ~
                            outputs.counts.values.eligible_ids ~ ' - (' ~
                            outputs.counts.values.eligible_ids ~ ' - $p)) |
                            length') | first }}"

                      - id: verify_gate
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ (outputs.verify_counts.values.still_count | number) > 0 }}"
                        then:
                          - id: resume_failed
                            type: io.kestra.plugin.core.execution.Fail
                            errorMessage: "Resume did not land: {{ outputs.verify_counts.values.still_count
                              }} targeted execution(s) still PAUSED ({{
                              outputs.verify_counts.values.still_ids }}).
                              required_pause_task may have held one back. The
                              answer may not match the pause's onResume inputs.
                              The API user may not be allowed to resume on that
                              namespace."
                        else:
                          - id: recovered
                            type: io.kestra.plugin.core.log.Log
                            message: "Answered and verified {{ outputs.counts.values.eligible_count }}
                              abandoned pause(s): {{
                              outputs.counts.values.eligible_ids }}. None of
                              them is PAUSED any more. Note: {{
                              outputs.approval.onResume.note ?? 'none' }}."
                    else:
                      - id: declined
                        type: io.kestra.plugin.core.log.Log
                        message: "Approval not granted, so no pause was answered. {{
                          outputs.counts.values.eligible_count }} execution(s)
                          are still PAUSED."
            else:
              - id: none_eligible
                type: io.kestra.plugin.core.log.Log
                message: "{{ outputs.counts.values.stale_count }} execution(s) are PAUSED past
                  the window but none is on allow_flows, so nothing was
                  answered. Held back: {{ outputs.counts.values.held_detail }}."
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: "Report only (auto_remediate={{ inputs.auto_remediate }} dry_run={{
              inputs.dry_run }}). Would answer {{
              outputs.counts.values.eligible_count }} of {{
              outputs.counts.values.stale_count }} stale pause(s): {{
              outputs.counts.values.eligible_detail }}."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No execution in namespace {{ inputs.namespace }} has been PAUSED over
          {{ inputs.min_paused_minutes }} minute(s). {{
          outputs.counts.values.all_count }} pause(s) are inside the grace
          window."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kestra stuck paused execution recovery guard failed on ' ~ inputs.kestra_url ~ ' (execution ' ~ execution.id ~ '). A resume may not have landed. The target may be unreachable. Check the logs.'} | toJson }}

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "25 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

An execution that hit a Pause waiting for a human answer, then never got one, sits PAUSED for ever. It holds its concurrency budget. Its downstream never runs. Because the execution is not FAILED no failure alert fires. Weeks later nobody remembers what it was waiting for. This flow finds those executions, reports them, then answers them under conditions you set.

It talks to the Kestra API of whichever instance you point kestra_url at. The default is the cluster the flow runs on, so it guards itself.

This blueprint was created by zkasuran.

Safety

Resuming a pause means answering an approval nobody answered. A blind resume can approve a destructive action that a human deliberately walked away from. This flow is for pauses abandoned by accident, never for approving work on somebody's behalf. The allowlist is what keeps those two apart.

allow_flows is empty by default, so out of the box nothing is ever resumed. You name the flows whose pauses you have decided are safe to answer, one flow id at a time. A paused execution of any other flow is reported in the logs, in Slack, then left alone.
min_paused_minutes is a grace window. A pause younger than the window is somebody's live decision, not an abandoned one, so it is never a candidate.
required_pause_task pins the task id the candidate must be waiting on. A candidate paused somewhere else is skipped, the re-verify then sees it still PAUSED, so the run fails rather than passing quietly.
resume_answer is the answer you wrote, sent verbatim as the pause's onResume payload. The flow never invents an approval value.
max_resumes refuses the batch rather than answering it when the pile is large. A large pile means approvals are not being answered at all, which is a process problem.
A resume runs only when all of these hold: auto_remediate true, dry_run false, the flow on allow_flows, the count at or under max_resumes, a human approving this flow's own Pause. Any missing step changes nothing.
After the resumes it queries the target again and fails the run if any targeted execution is still PAUSED.

How it works

window freezes the grace-window cutoff once, so every later comparison uses the same instant instead of drifting between tasks.
detect (io.kestra.plugin.kestra.executions.Query, read-only) fetches every PAUSED execution in the namespace with fetchType: FETCH. Its rows carry state.histories, which is where the moment an execution went PAUSED comes from. Execution start date is the wrong field here: a long run that paused a minute ago started hours ago.
paused_total (io.kestra.plugin.kestra.executions.Count, read-only) asks the same cluster for its own PAUSED count, so the detail query is checked against the target rather than trusted.
select and pick map each row to its PAUSED timestamp, drop this flow plus this execution, keep what is past the cutoff, then narrow that to allow_flows.
counts emits the ids, the stale detail, the allowlisted detail plus what the allowlist held back. summary logs all of it, which is the whole output in dry run.
route (io.kestra.plugin.core.flow.If):
- nothing stale: clean logs how many pauses are inside the window.
- stale found: alert posts to Slack.
  - report_only logs what would be answered unless auto_remediate is true with dry_run false.
  - eligible_gate stops at none_eligible when the allowlist leaves nothing.
  - ceiling fails the run rather than answering when more than max_resumes are eligible.
  - approval (io.kestra.plugin.core.flow.Pause) waits for a human. No answer changes nothing.
  - recover (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) reads each execution with detail, checks required_pause_task in task_gate, then answers it with io.kestra.plugin.kestra.executions.Resume carrying your resume_answer.
  - settle, then reverify queries PAUSED again and verify_counts intersects that with the ids targeted.
  - verify_gate fails the run when any targeted execution is still PAUSED. It logs recovered when none is.

What you get

A named list of the executions nobody answered, with the flow id plus how long each has been waiting, in the logs and in Slack.
A recovery that is bounded four ways: a grace window, a flow allowlist, an expected task id, a per-run ceiling.
An answer you control. The onResume payload is your JSON, not a hardcoded approval.
A re-query that fails the run when a resume did not land, so a silent no-op cannot pass as a fix.
A disabled hourly schedule that runs in report-only mode when you enable it.

Who it's for

Platform teams running Kestra with human-in-the-loop approvals. If any flow of yours pauses for a release sign-off, a data-quality sign-off or a cost sign-off, some of those pauses are already stale and nothing is telling you. Also for anyone who has watched a scheduled flow queue behind a pause from three weeks ago.

Why orchestrate this with Kestra

The detection, the approval gate plus the write all live in one versioned flow, against Kestra's own typed plugin rather than curl. The resume is itself gated by a Pause, so answering somebody else's approval needs an approval, which is the shape this problem deserves. Kestra keeps the full run history, so every recovery has a record of who approved it, what answer was sent, which executions moved. The schedule means you find the stale pauses rather than remembering to look.

Tested end to end

Run against Kestra 2.0.5 on a dedicated instance, never a shared one, with two flows pausing on named approval tasks.

Grace window, nothing to do: two pauses 30 seconds old with a 60 minute window, zero stale, clean logged.
Detection with identical inputs once one pause aged past a 1 minute window: one stale, named by flow id plus id.
dry_run=true: the stale pause named in the log plus Slack, both executions still PAUSED afterwards, confirmed by id through the target's own API.
auto_remediate=true, dry_run=false with an approved Pause: the allowlisted execution went PAUSED to SUCCESS, its onResume answer arrived as sent, the re-query found it gone.
Declined Pause: nothing resumed, target still PAUSED.
Allowlist exclusion: the non-allowlisted execution stayed PAUSED through every remediating run, checked by id each time.
Per-run ceiling: two eligible with max_resumes: 1 failed the run before the approval.
Re-verify failure: required_pause_task set to a task the candidate is not on left it PAUSED, the re-query caught it, the run FAILED.
Failure path: an unreachable kestra_url fired the errors: Slack block.

Prerequisites

Secrets

SLACK_WEBHOOK_URL: incoming webhook for the alert plus the failure notice.
KESTRA_PASSWORD: password for the API user on the target cluster. Default of the kestra_password input, so it is never written into the flow.

The API user needs read on executions in the namespace plus permission to resume them.

Inputs

kestra_url: target Kestra API base URL. Defaults to the local cluster.
namespace: the only namespace considered.
min_paused_minutes: grace window before a pause counts as abandoned. Default 60.
allow_flows: flow ids whose pauses may be answered. Empty by default, which resumes nothing.
required_pause_task: expected paused task id. Empty skips the task check.
resume_answer: JSON object sent as the onResume answer. Default {"approved": true}.
max_resumes: ceiling per run. Default 5.
kestra_user, kestra_password: API credentials.
auto_remediate, dry_run: the write gate. Defaults are report-only.

Quick start

Add SLACK_WEBHOOK_URL plus KESTRA_PASSWORD to your namespace secrets.
Run it as shipped with dry_run=true. You get the list of abandoned pauses, nothing changes.
Read the list. For each flow on it, decide whether its pause was abandoned by accident or walked away from on purpose. Only the accidental ones belong in allow_flows.
Put those flow ids in allow_flows, set required_pause_task to the approval task id, set resume_answer to the answer that flow's onResume expects.
Run with auto_remediate=true plus dry_run=false, then answer the approval pause with approved: true. Check the recovered log line.
Enable the hourly trigger once you are happy. It runs report-only, so it alerts without touching anything.

How to extend

Swap the Slack task for io.kestra.plugin.notifications.pagerduty or Opsgenie when a pause older than a day is an on-call matter.
Add a second window: alert at 60 minutes, only allow a resume past 24 hours, so a human has a day to answer before anything automated does.
Kill rather than answer. Point the loop at io.kestra.plugin.kestra.executions.Kill for pauses whose approval window has closed, where cancelling is safer than approving.
Record every recovery in a table with io.kestra.plugin.jdbc.postgresql.Query so the audit trail outlives the execution retention.
Widen the scope with io.kestra.plugin.kestra.namespaces.NamespacesWithFlows plus a loop over namespaces, keeping one allowlist per namespace.

Recover Kestra executions abandoned in PAUSED

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Recover Kestra executions abandoned in PAUSED

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

HashiCorp Consul Raft Leader Stability and Quorum Sentinel

Lint Commit Messages Against Conventional Commits Before Every Release

Kubernetes CoreDNS Request Latency Spike and Upstream RTT Sentinel

Audit the Operating System Inside Your Container Images for End of Life

Multi-Tenant CI/CD, Promote Kestra Flows Across Environments and Tenants Automatically

Scheduled Container Registry Vulnerability Audit with Trivy and Slack Alerts

More Core Blueprints

Promote Kestra flows from staging to production with approval, snapshot and rollback

Kestra Flow Failure Rate Gate

Kestra Flow Security, Governance and Linting Gate

Triage and Replay Failed Kestra Executions

Kestra instance retention policy with dry run, approval and audit CSV

Gate a Kestra upgrade against every live flow

New to Kestra?