Smoke test a Camunda BPMN deployment before you trust it

id: camunda-bpmn-deploy-smoke-gate
namespace: company.team
description: Deploy a bounded staging canary BPMN model, drive one uniquely
  correlated message and prove its exact process instance completed.
concurrency:
  limit: 1
  behavior: QUEUE
inputs:
  - id: runner_network
    type: STRING
    defaults: ''
    description: Optional Docker runner network for private staging endpoints. Empty
      uses the worker default.
  - id: camunda_rest_address
    type: STRING
    defaults: '{{ secret(''CAMUNDA_REST_ADDRESS'') }}'
    description: Dedicated staging Camunda 8 cluster REST API. Never point at a
      production process cluster.
  - id: camunda_username
    type: STRING
    defaults: '{{ secret(''CAMUNDA_USERNAME'') }}'
  - id: camunda_password
    type: STRING
    defaults: '{{ secret(''CAMUNDA_PASSWORD'') }}'
  - id: canary_process_id
    type: STRING
    defaults: canary_smoke
    description: Exact dedicated canary_ BPMN process ID. Restricts side effects and
      deployment scope.
  - id: bpmn_content
    type: STRING
    defaults: "<?xml version=\"1.0\" encoding=\"UTF-8\"?>

      <definitions xmlns=\"http://www.omg.org/spec/BPMN/20100524/MODEL\"
      xmlns:zeebe=\"http://camunda.org/schema/zeebe/1.0\"
      targetNamespace=\"http://example.org/canary\">

      \  <message id=\"smoke_message\"
      name=\"canary_smoke_continue\"><extensionElements><zeebe:subscription
      correlationKey=\"=canaryCorrelationKey\" /></extensionElements></message>

      \  <process id=\"canary_smoke\" isExecutable=\"true\">

      \    <startEvent id=\"start\"><outgoing>flow1</outgoing></startEvent>

      \    <intermediateCatchEvent
      id=\"wait\"><incoming>flow1</incoming><outgoing>flow2</outgoing><messageE\
      ventDefinition messageRef=\"smoke_message\" /></intermediateCatchEvent>

      \    <endEvent id=\"end\"><incoming>flow2</incoming></endEvent>

      \    <sequenceFlow id=\"flow1\" sourceRef=\"start\" targetRef=\"wait\" />

      \    <sequenceFlow id=\"flow2\" sourceRef=\"wait\" targetRef=\"end\" />

      \  </process>

      </definitions>"
    description: At most 64 KiB. One plain start, one message catch and one end.
      Task, call, timer, signal and script elements are refused.
  - id: message_name
    type: STRING
    defaults: canary_smoke_continue
    description: Expected catch message name. Must match the staged BPMN.
  - id: receipt_key
    type: STRING
    defaults: camunda_canary_smoke
    description: Stable dedicated KV receipt key for this cluster/process. Change it
      only to force a reviewed new test.
  - id: verify_seconds
    type: INT
    defaults: 30
    min: 10
    max: 60
  - id: dry_run
    type: BOOL
    defaults: true
  - id: auto_remediate
    type: BOOL
    defaults: false
  - id: slack_webhook_url
    type: STRING
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"
variables:
  reader_script: "import
    base64,hashlib,json,os,pathlib,time,urllib.request,urllib.parse,xml.etree.E\
    lementTree as ET

    ROOT=pathlib.Path('.')

    BASE=os.environ['BASE'].rstrip('/')

    USER=os.environ['USER'];PASSWORD=os.environ['PASSWORD']

    HEADERS={'Content-Type':'application/json'}

    if USER or PASSWORD:

    \    if not USER or not PASSWORD: raise RuntimeError('Both basic
    authentication fields are required')

    \    HEADERS['Authorization']='Basic
    '+base64.b64encode((USER+':'+PASSWORD).encode()).decode()

    def api(path,body=None):

    \    req=urllib.request.Request(BASE+path,data=(json.dumps(body).encode() if
    body is not None else None),headers=HEADERS)

    \    with urllib.request.urlopen(req,timeout=10) as r: return json.load(r)

    def save(name,data): (ROOT/name).write_text(json.dumps(data,sort_keys=True))

    def marker(data):print('::'+json.dumps({'outputs':data})+'::',flush=True)

    def definitions(pid):

    \    d=api('/v2/process-definitions/search',{'filter':{'processDefinitionId\
    ':pid},'page':{'limit':51}})

    \    if d['page'].get('hasMoreTotalItems') or len(d['items'])>50:raise
    RuntimeError('Canary has more than 50 retained versions. Retire versions
    through reviewed engine operations')

    \    return
    sorted([{'key':x['processDefinitionKey'],'version':x['version'],'state':x['\
    state']} for x in d['items']],key=lambda x:x['version'])

    mode=os.environ['MODE']

    if mode in ['plan','recheck']:

    \    raw=(ROOT/'model.bpmn').read_bytes()

    \    if len(raw)>65536:raise RuntimeError('BPMN exceeds 64 KiB limit')

    \    tree=ET.fromstring(raw)

    \    B='{http://www.omg.org/spec/BPMN/20100524/MODEL}'

    \    process=tree.findall(B+'process')

    \    if len(process)!=1 or process[0].get('isExecutable')!='true':raise
    RuntimeError('Exactly one executable process is required')

    \    pid=process[0].get('id','')

    \    if pid!=os.environ['PROCESS_ID'] or not pid.startswith('canary_'):raise
    RuntimeError('Only the exact canary_ process ID is allowed')

    \    allowed={'definitions','process','message','extensionElements','subscr\
    iption','sequenceFlow','startEvent','intermediateCatchEvent','messageEventD\
    efinition','endEvent','incoming','outgoing','documentation'}

    \    for node in tree.iter():

    \        if node.tag.split('}')[-1] not in allowed:raise
    RuntimeError('Unsupported side-effect-capable BPMN element: '+node.tag)

    \    starts=process[0].findall(B+'startEvent');ends=process[0].findall(B+'e\
    ndEvent');waits=process[0].findall(B+'intermediateCatchEvent')

    \    if len(starts)!=1 or len(ends)!=1 or len(waits)!=1 or
    list(starts[0])!=[starts[0].find(B+'outgoing')]:raise RuntimeError('Canary
    needs one plain start, one message catch and one end')

    \    if waits[0].find(B+'messageEventDefinition') is None:raise
    RuntimeError('Canary catch must be a message')

    \    messages=tree.findall(B+'message')

    \    flows=process[0].findall(B+'sequenceFlow')

    \    if len(messages)!=1 or len(flows)!=2:raise RuntimeError('Canary needs
    exactly one message and two sequence flows')

    \    edges={(x.get('sourceRef'),x.get('targetRef')) for x in flows}

    \    if
    edges!={(starts[0].get('id'),waits[0].get('id')),(waits[0].get('id'),ends[0\
    ].get('id'))}:raise RuntimeError('Canary must follow one linear
    start-message-end path')

    \    if any(x.tag.split('}')[-1].endswith('EventDefinition') for x in
    starts[0]):raise RuntimeError('Only a plain start is allowed')

    \    topology=api('/v2/topology')

    \    if not topology.get('clusterId'):raise RuntimeError('Camunda cluster ID
    is required')

    \    result={'cluster_id':topology['clusterId'],'process_id':pid,'bpmn_sha2\
    56':hashlib.sha256(raw).hexdigest(),'definitions':definitions(pid),'endpoin\
    t':BASE}

    \    if mode=='recheck':

    \        original=json.loads((ROOT/'approved-plan.json').read_text())

    \        if original!=result:raise RuntimeError('Approved cluster, BPMN or
    canary definition versions changed')

    \        fresh=(ROOT/'active.json').read_text().strip()

    \        approved=(ROOT/'approved-active.json').read_text().strip()

    \        if fresh!=approved:raise RuntimeError('Residual canary key changed
    during approval. Re-plan instead of writing')

    \        if fresh:

    \            held=json.loads(fresh)

    \            still=api('/v2/process-instances/search',{'filter':{'processIn\
    stanceKey':str(held['instance_key'])},'page':{'limit':2}})

    \            if len(still['items'])!=1 or still['items'][0]['state'] not in
    ['COMPLETED','TERMINATED']:raise RuntimeError('Residual canary became active
    during approval')

    \    save('plan.json',result);marker(result)

    elif mode=='receipt':

    \    plan=json.loads((ROOT/'approved-plan.json').read_text())

    \    receipt_raw=(ROOT/'receipt.json').read_text().strip()

    \    active_raw=(ROOT/'active.json').read_text().strip()

    \    receipt=json.loads(receipt_raw) if receipt_raw else {}

    \    active=json.loads(active_raw) if active_raw else {}

    \    if active:

    \        if active.get('cluster_id')!=plan['cluster_id']:raise
    RuntimeError('Residual canary receipt belongs to another cluster. Inspect it
    before replacement')

    \        previous=api('/v2/process-instances/search',{'filter':{'processIns\
    tanceKey':str(active['instance_key'])},'page':{'limit':2}})

    \        if len(previous['items'])!=1 or previous['items'][0]['state'] not
    in ['COMPLETED','TERMINATED']:

    \            raise RuntimeError('Residual created canary is still active or
    unknown. Reconcile its saved key before creating another')

    \        if
    str(previous['items'][0]['processDefinitionKey'])!=str(active['definition_k\
    ey']):raise RuntimeError('Residual instance identity mismatch')

    \    matching=bool(receipt and receipt.get('cluster_id')==plan['cluster_id']
    and receipt.get('bpmn_sha256')==plan['bpmn_sha256'] and
    receipt.get('process_id')==plan['process_id'] and
    receipt.get('message_name')==os.environ['MESSAGE_NAME'])

    \    if matching:

    \        previous=api('/v2/process-instances/search',{'filter':{'processIns\
    tanceKey':str(receipt['instance_key'])},'page':{'limit':2}})

    \        definitions_now=api('/v2/process-definitions/search',{'filter':{'p\
    rocessDefinitionKey':str(receipt['definition_key'])},'page':{'limit':2}})

    \        correlations=api('/v2/correlated-message-subscriptions/search',{'f\
    ilter':{'processInstanceKey':str(receipt['instance_key'])},'page':{'limit':\
    2}})

    \        matching=bool(len(previous['items'])==1 and
    previous['items'][0]['state']=='COMPLETED' and
    str(previous['items'][0]['processDefinitionKey'])==str(receipt['definition_\
    key']) and len(definitions_now['items'])==1 and
    definitions_now['items'][0]['state']=='ACTIVE' and
    any(str(x['messageKey'])==str(receipt['message_key']) and
    x['correlationKey']==receipt['execution_id'] for x in
    correlations['items']))

    \        save('receipt-check.json',{'receipt':receipt,'instances':previous,\
    'definitions':definitions_now,'correlations':correlations,'matching':matchi\
    ng})

    \    marker({'needs_test':not matching,'residual_terminal':bool(active)})

    elif mode in ['subscription','outcome','cleanup_inspect','cleanup_verify']:

    \    key=os.environ['INSTANCE'];definition=os.environ['DEFINITION'];correla\
    tion=os.environ['CORRELATION']

    \    if not key or not definition:raise RuntimeError('Owned instance and
    definition keys required')

    \    deadline=time.monotonic()+int(os.environ['VERIFY_SECONDS'])

    \    while True:

    \        d=api('/v2/process-instances/search',{'filter':{'processInstanceKe\
    y':key},'page':{'limit':2}})

    \        if len(d['items'])>1:raise RuntimeError('Instance key returned
    multiple instances')

    \        instance=d['items'][0] if d['items'] else None

    \        if instance and (str(instance['processDefinitionKey'])!=definition
    or instance['processDefinitionId']!=os.environ['PROCESS_ID']):raise
    RuntimeError('Owned instance identity changed')

    \        if mode=='subscription':

    \            subs=api('/v2/message-subscriptions/search',{'filter':{'proces\
    sInstanceKey':key},'page':{'limit':2}})

    \            good=[x for x in subs['items'] if
    str(x['processDefinitionKey'])==definition and
    x['correlationKey']==correlation and
    x['messageName']==os.environ['MESSAGE_NAME'] and
    x['messageSubscriptionState']=='CREATED']

    \            if len(good)==1 and instance and instance['state']=='ACTIVE':

    \                save('subscription.json',{'instance':instance,'subscriptio\
    ns':subs});marker({'subscription_ready':True});break

    \        elif mode=='outcome':

    \            correlations=api('/v2/correlated-message-subscriptions/search'\
    ,{'filter':{'processInstanceKey':key},'page':{'limit':2}})

    \            good=[x for x in correlations['items'] if
    str(x['processDefinitionKey'])==definition and
    x['correlationKey']==correlation and
    x['messageName']==os.environ['MESSAGE_NAME'] and
    str(x['messageKey'])==os.environ['MESSAGE_KEY']]

    \            result={'instance':instance,'correlations':correlations}

    \            if instance and instance['state']=='COMPLETED' and not
    instance['hasIncident'] and len(good)==1:

    \                save('outcome.json',result);marker({'completed':True,'inst\
    ance_key':key,'definition_key':definition});break

    \        elif mode=='cleanup_inspect':

    \            result={'instance':instance,'should_cancel':bool(instance and
    instance['state']=='ACTIVE')}

    \            save('cleanup.json',result);marker({'should_cancel':result['sh\
    ould_cancel']});break

    \        else:

    \            if instance and
    instance['state']=='TERMINATED':save('cleanup.json',{'instance':instance});\
    marker({'terminated':True});break

    \        if time.monotonic()>=deadline:

    \            save('unresolved.json',{'instance':instance,'subscriptions':su\
    bs if mode=='subscription' else None,'correlations':correlations if
    mode=='outcome' else None})

    \            raise RuntimeError('Owned canary did not reach required
    '+mode+' state within bound')

    \        time.sleep(2)

    else:raise RuntimeError('Unknown read mode')\n"
tasks:
  - id: plan
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.12-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: '{{ inputs.runner_network }}'
    timeout: PT90S
    env:
      BASE: '{{ inputs.camunda_rest_address }}'
      USER: '{{ inputs.camunda_username }}'
      PASSWORD: '{{ inputs.camunda_password }}'
      MODE: plan
      MESSAGE_NAME: '{{ inputs.message_name }}'
      PROCESS_ID: '{{ inputs.canary_process_id }}'
      VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
    script: '{{ vars.reader_script }}'
    outputFiles:
      - '*.json'
    inputFiles:
      model.bpmn: '{{ inputs.bpmn_content }}'
  - id: read_receipt
    type: io.kestra.plugin.core.kv.Get
    key: '{{ inputs.receipt_key }}'
    errorOnMissing: false
  - id: read_active
    type: io.kestra.plugin.core.kv.Get
    key: '{{ inputs.receipt_key }}_active'
    errorOnMissing: false
  - id: receipt_check
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.12-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: '{{ inputs.runner_network }}'
    timeout: PT90S
    env:
      BASE: '{{ inputs.camunda_rest_address }}'
      USER: '{{ inputs.camunda_username }}'
      PASSWORD: '{{ inputs.camunda_password }}'
      MODE: receipt
      MESSAGE_NAME: '{{ inputs.message_name }}'
      PROCESS_ID: '{{ inputs.canary_process_id }}'
      VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
    script: '{{ vars.reader_script }}'
    outputFiles:
      - '*.json'
    inputFiles:
      approved-plan.json: '{{ read(outputs.plan.outputFiles["plan.json"]) }}'
      receipt.json: '{{ outputs.read_receipt.value ?? '''' }}'
      active.json: '{{ outputs.read_active.value ?? '''' }}'
  - id: needs_test
    type: io.kestra.plugin.core.flow.If
    condition: '{{ outputs.receipt_check.vars.needs_test }}'
    then:
      - id: report
        type: io.kestra.plugin.core.log.Log
        message: New or changed staging canary requires smoke proof. Planned cluster {{
          outputs.plan.vars.cluster_id }} model SHA {{
          outputs.plan.vars.bpmn_sha256 }}. Existing versions {{
          outputs.plan.vars.definitions | toJson }}.
      - id: mode
        type: io.kestra.plugin.core.flow.If
        condition: '{{ inputs.auto_remediate and not inputs.dry_run }}'
        then:
          - id: approval
            type: io.kestra.plugin.core.flow.Pause
            pauseDuration: PT30M
            behavior: FAIL
            description: Deploy one retained staging canary version and create one isolated
              message-driven canary. Completed history and deployment remain.
              Approve only after reading model and exact staging endpoint.
            onResume:
              - id: approved
                type: BOOL
                required: false
          - id: approved
            type: io.kestra.plugin.core.flow.If
            condition: '{{ outputs.approval.onResume.approved ?? false }}'
            then:
              - id: reread_active
                type: io.kestra.plugin.core.kv.Get
                key: '{{ inputs.receipt_key }}_active'
                errorOnMissing: false
              - id: recheck
                type: io.kestra.plugin.scripts.python.Script
                containerImage: python:3.12-slim
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: '{{ inputs.runner_network }}'
                timeout: PT90S
                env:
                  BASE: '{{ inputs.camunda_rest_address }}'
                  USER: '{{ inputs.camunda_username }}'
                  PASSWORD: '{{ inputs.camunda_password }}'
                  MODE: recheck
                  MESSAGE_NAME: '{{ inputs.message_name }}'
                  PROCESS_ID: '{{ inputs.canary_process_id }}'
                  VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
                script: '{{ vars.reader_script }}'
                outputFiles:
                  - '*.json'
                inputFiles:
                  model.bpmn: '{{ inputs.bpmn_content }}'
                  approved-plan.json: '{{ read(outputs.plan.outputFiles["plan.json"]) }}'
                  active.json: '{{ outputs.reread_active.value ?? '''' }}'
                  approved-active.json: '{{ outputs.read_active.value ?? '''' }}'
              - id: deploy
                type: io.kestra.plugin.camunda.Deploy
                restAddress: '{{ inputs.camunda_rest_address }}'
                username: '{{ inputs.camunda_username }}'
                password: '{{ inputs.camunda_password }}'
                timeout: PT30S
                resources:
                  canary.bpmn: '{{ inputs.bpmn_content }}'
              - id: deployed_scope
                type: io.kestra.plugin.core.execution.Assert
                conditions:
                  - '{{ (outputs.deploy.processes | length) == 1 and
                    outputs.deploy.processes[0].bpmnProcessId ==
                    inputs.canary_process_id }}'
                errorMessage: Native deployment did not return the exact single canary process.
              - id: start_canary
                type: io.kestra.plugin.camunda.CreateProcessInstance
                restAddress: '{{ inputs.camunda_rest_address }}'
                username: '{{ inputs.camunda_username }}'
                password: '{{ inputs.camunda_password }}'
                timeout: PT30S
                processDefinitionKey: '{{ outputs.deploy.processes[0].processDefinitionKey }}'
                awaitCompletion: false
                requestTimeout: PT10S
                variables:
                  canaryCorrelationKey: '{{ execution.id }}'
                  kestraExecutionId: '{{ execution.id }}'
              - id: save_instance
                type: io.kestra.plugin.core.kv.Set
                key: '{{ inputs.receipt_key }}_active'
                value: '{{ {''instance_key'': outputs.start_canary.processInstanceKey,
                  ''definition_key'':
                  outputs.deploy.processes[0].processDefinitionKey,
                  ''execution_id'': execution.id, ''cluster_id'':
                  outputs.plan.vars.cluster_id} | toJson }}'
              - id: waiting_subscription
                type: io.kestra.plugin.scripts.python.Script
                containerImage: python:3.12-slim
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: '{{ inputs.runner_network }}'
                timeout: PT90S
                env:
                  BASE: '{{ inputs.camunda_rest_address }}'
                  USER: '{{ inputs.camunda_username }}'
                  PASSWORD: '{{ inputs.camunda_password }}'
                  MODE: subscription
                  MESSAGE_NAME: '{{ inputs.message_name }}'
                  PROCESS_ID: '{{ inputs.canary_process_id }}'
                  VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
                  INSTANCE: '{{ outputs.start_canary.processInstanceKey }}'
                  DEFINITION: '{{ outputs.deploy.processes[0].processDefinitionKey }}'
                  CORRELATION: '{{ execution.id }}'
                  MESSAGE_KEY: '{{ outputs.publish.messageKey ?? '''' }}'
                script: '{{ vars.reader_script }}'
                outputFiles:
                  - '*.json'
              - id: publish
                type: io.kestra.plugin.camunda.PublishMessage
                restAddress: '{{ inputs.camunda_rest_address }}'
                username: '{{ inputs.camunda_username }}'
                password: '{{ inputs.camunda_password }}'
                timeout: PT30S
                messageName: '{{ inputs.message_name }}'
                correlationKey: '{{ execution.id }}'
                messageId: '{{ execution.id }}'
                timeToLive: PT0S
                variables:
                  canaryDeliveredBy: '{{ execution.id }}'
              - id: prove_completion
                type: io.kestra.plugin.scripts.python.Script
                containerImage: python:3.12-slim
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: '{{ inputs.runner_network }}'
                timeout: PT90S
                env:
                  BASE: '{{ inputs.camunda_rest_address }}'
                  USER: '{{ inputs.camunda_username }}'
                  PASSWORD: '{{ inputs.camunda_password }}'
                  MODE: outcome
                  MESSAGE_NAME: '{{ inputs.message_name }}'
                  PROCESS_ID: '{{ inputs.canary_process_id }}'
                  VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
                  INSTANCE: '{{ outputs.start_canary.processInstanceKey }}'
                  DEFINITION: '{{ outputs.deploy.processes[0].processDefinitionKey }}'
                  CORRELATION: '{{ execution.id }}'
                  MESSAGE_KEY: '{{ outputs.publish.messageKey ?? '''' }}'
                script: '{{ vars.reader_script }}'
                outputFiles:
                  - '*.json'
              - id: save_receipt
                type: io.kestra.plugin.core.kv.Set
                key: '{{ inputs.receipt_key }}'
                value: '{{ {''cluster_id'': outputs.plan.vars.cluster_id, ''bpmn_sha256'':
                  outputs.plan.vars.bpmn_sha256, ''process_id'':
                  inputs.canary_process_id, ''message_name'':
                  inputs.message_name, ''instance_key'':
                  outputs.start_canary.processInstanceKey, ''definition_key'':
                  outputs.deploy.processes[0].processDefinitionKey,
                  ''execution_id'': execution.id, ''message_key'':
                  outputs.publish.messageKey, ''completed'': true} | toJson }}'
              - id: clear_active
                type: io.kestra.plugin.core.kv.Delete
                key: '{{ inputs.receipt_key }}_active'
              - id: completed
                type: io.kestra.plugin.core.log.Log
                message: Exact staging canary {{ outputs.start_canary.processInstanceKey }}
                  completed after message {{ outputs.publish.messageKey }} on
                  deployed definition {{
                  outputs.deploy.processes[0].processDefinitionKey }}.
                  Deployment and completed history remain.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: Approval declined. No BPMN deployment, process instance or message was
                  created.
        else:
          - id: dry_run
            type: io.kestra.plugin.core.log.Log
            message: Read-only plan. No deployment, instance or message created.
    else:
      - id: healthy_receipt
        type: io.kestra.plugin.core.log.Log
        message: Matching completed smoke receipt exists. No duplicate deployment or
          canary was created.
errors:
  - id: cleanup_owned
    type: io.kestra.plugin.core.flow.If
    condition: '{{ outputs.start_canary.processInstanceKey is defined }}'
    then:
      - id: inspect_owned
        type: io.kestra.plugin.scripts.python.Script
        containerImage: python:3.12-slim
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: '{{ inputs.runner_network }}'
        timeout: PT90S
        env:
          BASE: '{{ inputs.camunda_rest_address }}'
          USER: '{{ inputs.camunda_username }}'
          PASSWORD: '{{ inputs.camunda_password }}'
          MODE: cleanup_inspect
          MESSAGE_NAME: '{{ inputs.message_name }}'
          PROCESS_ID: '{{ inputs.canary_process_id }}'
          VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
          INSTANCE: '{{ outputs.start_canary.processInstanceKey }}'
          DEFINITION: '{{ outputs.deploy.processes[0].processDefinitionKey }}'
          CORRELATION: '{{ execution.id }}'
          MESSAGE_KEY: '{{ outputs.publish.messageKey ?? '''' }}'
        script: '{{ vars.reader_script }}'
        outputFiles:
          - '*.json'
      - id: cancel_if_active
        type: io.kestra.plugin.core.flow.If
        condition: '{{ outputs.inspect_owned.vars.should_cancel ?? false }}'
        then:
          - id: cancel_canary
            type: io.kestra.plugin.camunda.CancelProcessInstance
            restAddress: '{{ inputs.camunda_rest_address }}'
            username: '{{ inputs.camunda_username }}'
            password: '{{ inputs.camunda_password }}'
            timeout: PT30S
            processInstanceKey: '{{ outputs.start_canary.processInstanceKey }}'
          - id: verify_cleanup
            type: io.kestra.plugin.scripts.python.Script
            containerImage: python:3.12-slim
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              networkMode: '{{ inputs.runner_network }}'
            timeout: PT90S
            env:
              BASE: '{{ inputs.camunda_rest_address }}'
              USER: '{{ inputs.camunda_username }}'
              PASSWORD: '{{ inputs.camunda_password }}'
              MODE: cleanup_verify
              MESSAGE_NAME: '{{ inputs.message_name }}'
              PROCESS_ID: '{{ inputs.canary_process_id }}'
              VERIFY_SECONDS: '{{ inputs.verify_seconds }}'
              INSTANCE: '{{ outputs.start_canary.processInstanceKey }}'
              DEFINITION: '{{ outputs.deploy.processes[0].processDefinitionKey }}'
              CORRELATION: '{{ execution.id }}'
              MESSAGE_KEY: '{{ outputs.publish.messageKey ?? '''' }}'
            script: '{{ vars.reader_script }}'
            outputFiles:
              - '*.json'
  - id: failure_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: '{{ inputs.slack_webhook_url }}'
    messageText: Camunda staging deployment smoke failed in {{ execution.id }}.
      Inspect native keys and cleanup evidence. Retained canary deployments are
      not rolled back. A timed-out create can leave an instance whose key was
      not returned; investigate the unique execution correlation before
      retrying.
triggers:
  - id: staging_smoke_plan
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 7 * * *
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false