Governed Action Executor with Human Approval Gate

id: mcp-governed-action-executor
namespace: company.team

description: |
  Executor for governed actions. Not an MCP tool: it is started by mcp-governed-action-request,
  and can also be started by hand or by its webhook with a key. It only acts on a request row
  that is in status `claimed`, so a run with an unknown key, or a key that was already handled,
  does nothing.

inputs:
  - id: key
    type: STRING
    # Not required: a webhook run gets the key from the request body instead.
    required: false
    validator: '^[A-Za-z0-9_-]{1,64}$'
    description: Request key of a row created by mcp-governed-action-request.

variables:
  # Placeholders: replace with your own values.
  jdbc_url: jdbc:postgresql://demo-postgres:5432/demo
  jdbc_user: governed_actions
  kestra_url: http://localhost:8080
  # Environments that need a human approval. Fixed list, never an input.
  approval_environments: production
  # Fixed map: the schema name is never taken from free text.
  schemas:
    staging: staging
    production: production

tasks:
  # The key comes from the input, or from {"key": "..."} in a webhook body. Never from anything else.
  - id: resolve_key
    type: io.kestra.plugin.core.output.OutputValues
    values:
      key: "{{ inputs.key ?? trigger.body.key ?? '' }}"

  - id: claim_request
    # Webhook bodies are not validated by Kestra, so the pattern is also checked in SQL: an invalid key matches no row.
    description: Atomically take a claimed request. Zero rows means nothing to do.
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ vars.jdbc_url }}"
    username: "{{ vars.jdbc_user }}"
    password: "{{ secret('PG_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      UPDATE public.governed_requests
      SET status = CASE WHEN environment = ANY(string_to_array(CAST(:approval_environments AS text), ',')) THEN 'pending' ELSE 'running' END,
          executor_execution_id = CAST(:exec AS text),
          updated_at = now()
      WHERE request_key = CAST(:key AS text) AND status = 'claimed'
        AND CAST(:key AS text) ~ '^[A-Za-z0-9_-]{1,64}$'
        AND (executor_execution_id IS NULL OR executor_execution_id = CAST(:exec AS text))
      RETURNING environment, days, status
    parameters:
      key: "{{ outputs.resolve_key.values.key }}"
      exec: "{{ execution.id }}"
      approval_environments: "{{ vars.approval_environments }}"

  - id: has_claim
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.claim_request.size ?? 0) > 0 }}"
    then:
      - id: notify_approvers
        type: io.kestra.plugin.core.flow.Sequential
        runIf: "{{ outputs.claim_request.row.status == 'pending' }}"
        # A failed notification must never block the approval.
        allowFailure: true
        tasks:
          - id: notify_slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            retry:
              type: constant
              interval: PT10S
              maxAttempts: 3
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            # toJson escapes the whole text, so no input can break the JSON string.
            payload: |
              {
                "text": {{ ('Approval needed: ' ~ outputs.claim_request.row.environment ~ ' purge of rows older than ' ~ outputs.claim_request.row.days ~ ' days (request ' ~ outputs.resolve_key.values.key ~ '). Approve or deny: ' ~ vars.kestra_url ~ '/ui/' ~ flow.tenantId ~ '/executions/' ~ flow.namespace ~ '/' ~ flow.id ~ '/' ~ execution.id) | toJson }}
              }
        errors:
          - id: record_unnotified
            type: io.kestra.plugin.jdbc.postgresql.Query
            url: "{{ vars.jdbc_url }}"
            username: "{{ vars.jdbc_user }}"
            password: "{{ secret('PG_PASSWORD') }}"
            sql: |
              UPDATE public.governed_requests SET status = 'pending_unnotified', updated_at = now()
              WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text) AND status = 'pending'
            parameters:
              key: "{{ outputs.resolve_key.values.key }}"
              exec: "{{ execution.id }}"

      - id: wait_for_approval
        type: io.kestra.plugin.core.flow.Pause
        runIf: "{{ outputs.claim_request.row.status == 'pending' }}"
        # Literal values on purpose: templated pause settings are ignored in Kestra 2.0.4.
        # FAIL makes the `errors` block run on timeout; WARN and RESUME would run the next task.
        pauseDuration: PT1H
        behavior: FAIL
        onResume:
          - id: decision
            type: SELECT
            required: true
            defaults: deny
            values:
              - approve
              - deny
          - id: approver
            type: STRING
            required: false
            description: Self-asserted name of the approver.

      - id: label_execution
        type: io.kestra.plugin.core.execution.Labels
        allowFailure: true
        labels:
          # Label values cannot be empty: `default` also replaces an empty string.
          environment: "{{ outputs.claim_request.row.environment }}"
          decision: "{{ outputs.claim_request.row.status == 'pending' ?
            (outputs.wait_for_approval.onResume.decision | default('deny')) :
            'auto' }}"
          decided_by: "{{ outputs.claim_request.row.status == 'pending' ?
            (outputs.wait_for_approval.onResume.approver | default('unknown')) :
            'auto' }}"

      # The Pause is NOT the gate: resuming with a wrong or missing field proceeds with defaults.
      # Only an explicit decision == approve (or no approval required) runs the action.
      - id: authorization_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.claim_request.row.status == 'running' or
          (outputs.wait_for_approval.onResume.decision ?? 'deny') == 'approve'
          }}"
        then:
          - id: record_running
            type: io.kestra.plugin.jdbc.postgresql.Query
            url: "{{ vars.jdbc_url }}"
            username: "{{ vars.jdbc_user }}"
            password: "{{ secret('PG_PASSWORD') }}"
            sql: |
              UPDATE public.governed_requests
              SET status = 'running', decided_by = CAST(:decided_by AS text), updated_at = now()
              WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text)
                AND status IN ('pending', 'pending_unnotified', 'running')
            parameters:
              key: "{{ outputs.resolve_key.values.key }}"
              exec: "{{ execution.id }}"
              decided_by: "{{ outputs.claim_request.row.status == 'pending' ?
                (outputs.wait_for_approval.onResume.approver |
                default('unknown')) : 'auto' }}"

          # ONE statement: the DELETE and the `executed` update commit or roll back together.
          # Schema comes from the fixed map, days from the claimed row. Both the DELETE and the UPDATE
          # require the row to be `running` for this execution.
          - id: purge_and_record
            type: io.kestra.plugin.jdbc.postgresql.Query
            url: "{{ vars.jdbc_url }}"
            username: "{{ vars.jdbc_user }}"
            password: "{{ secret('PG_PASSWORD') }}"
            fetchType: FETCH_ONE
            sql: |
              WITH purged AS (
                DELETE FROM {{ vars.schemas[outputs.claim_request.row.environment] }}.demo_events
                WHERE created_at < now() - make_interval(days => CAST(:days AS integer))
                  AND EXISTS (
                    SELECT 1 FROM public.governed_requests
                    WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text) AND status = 'running'
                  )
                RETURNING 1
              )
              UPDATE public.governed_requests
              SET status = 'executed', updated_at = now()
              WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text) AND status = 'running'
              RETURNING (SELECT count(*) FROM purged) AS deleted_rows
            parameters:
              days: "{{ outputs.claim_request.row.days }}"
              key: "{{ outputs.resolve_key.values.key }}"
              exec: "{{ execution.id }}"

          # Zero rows means the row was no longer `running` for this execution (for example a Restart after
          # a recorded failure). Nothing was deleted: fail so the execution does not look successful.
          - id: fail_if_not_executed
            type: io.kestra.plugin.core.execution.Fail
            runIf: "{{ (outputs.purge_and_record.size ?? 0) == 0 }}"
            errorMessage: "Request {{ outputs.resolve_key.values.key }} is not running for
              this execution. Nothing was deleted."
        else:
          - id: record_rejected
            type: io.kestra.plugin.jdbc.postgresql.Query
            url: "{{ vars.jdbc_url }}"
            username: "{{ vars.jdbc_user }}"
            password: "{{ secret('PG_PASSWORD') }}"
            sql: |
              UPDATE public.governed_requests
              SET status = 'rejected', decided_by = CAST(:decided_by AS text), updated_at = now()
              WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text)
                AND status IN ('pending', 'pending_unnotified')
            parameters:
              key: "{{ outputs.resolve_key.values.key }}"
              exec: "{{ execution.id }}"
              decided_by: "{{ outputs.wait_for_approval.onResume.approver | default('unknown')
                }}"
    else:
      - id: nothing_to_do
        type: io.kestra.plugin.core.log.Log
        message: "Nothing to do: the key is empty or invalid, unknown, or its request is
          not in status claimed."

errors:
  # Pending: not approved within 1 hour, or the executor failed before the purge started. Running: the purge failed.
  - id: record_expired
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ vars.jdbc_url }}"
    username: "{{ vars.jdbc_user }}"
    password: "{{ secret('PG_PASSWORD') }}"
    sql: |
      UPDATE public.governed_requests SET status = 'expired', updated_at = now()
      WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text)
        AND status IN ('pending', 'pending_unnotified')
    parameters:
      key: "{{ outputs.resolve_key.values.key ?? '' }}"
      exec: "{{ execution.id }}"

  - id: record_failed
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ vars.jdbc_url }}"
    username: "{{ vars.jdbc_user }}"
    password: "{{ secret('PG_PASSWORD') }}"
    sql: |
      UPDATE public.governed_requests SET status = 'failed', updated_at = now()
      WHERE request_key = CAST(:key AS text) AND executor_execution_id = CAST(:exec AS text)
        AND status = 'running'
    parameters:
      key: "{{ outputs.resolve_key.values.key ?? '' }}"
      exec: "{{ execution.id }}"

triggers:
  - id: webhook
    type: io.kestra.plugin.core.trigger.Webhook
    # The webhook needs no Kestra credentials: this key is its only protection. Anyone who knows it can start
    # this flow with a body like {"key": "<request key>"}. It cannot bypass approval, but anyone with the key can
    # take over a `claimed` request, so the id returned to the agent may then point to a no-op execution.
    key: "{{ secret('GOVERNED_EXECUTOR_WEBHOOK_KEY') }}"