Failed to use encrypted output from Subflow

[danlsgiga]

Describe the issue

After migrating from v1.2.9 to v1.3.2 our flow is failing with

class io.kestra.core.models.flows.Output cannot be cast to class io.kestra.core.models.flows.input.SecretInput (io.kestra.core.models.flows.Output and io.kestra.core.models.flows.input.SecretInput are in unnamed module of loader 'app')

Flows to reproduce

id: flow_secret_output_issue
namespace: test

tasks:
  - id: get_secret
    type: io.kestra.plugin.core.flow.Subflow
    namespace: test
    flowId: get_secret
    inputs:
      path: /secret/kestra
    wait: true
    transmitFailed: true
  
  - id: get_token
    type: io.kestra.plugin.scripts.shell.Commands
    containerImage: ubuntu
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker      
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 10
      warningOnRetry: false        
    commands:
      - >
        echo -n '{"SECRET": "ASDF"}'
        > {{ execution.id }}
    outputFiles:
      - "{{ execution.id }}"

  - id: kv_set
    type: io.kestra.plugin.core.kv.Set
    key: "{{ execution.id }}"
    value: "{{ read(outputs.get_token.outputFiles[execution.id]) }}"
    kvType: STRING
    ttl: PT15S
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 10
      warningOnRetry: false

outputs:
  - id: token
    type: SECRET
    value: "{{ kv(execution.id) }}"

afterExecution:
  - id: delete_storage_file
    type: io.kestra.plugin.core.storage.Delete
    uri: "{{ outputs.get_token.outputFiles[execution.id] }}"
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 10
      warningOnRetry: false

id: get_secret
namespace: test

inputs:
  - id: token
    type: SECRET    
    displayName: Token
    required: false
  - id: path
    type: STRING
    displayName: Path

tasks:
  - id: get_secret
    type: io.kestra.plugin.scripts.shell.Commands
    containerImage: ubuntu
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker      
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 10
      warningOnRetry: false    
    commands:      
      - >
        echo '{"SECRET": "asdf"}'
        > {{ execution.id }}
    outputFiles:
      - "{{ execution.id }}"

  - id: kv_set
    type: io.kestra.plugin.core.kv.Set
    key: "{{ execution.id }}"
    value: "{{ read(outputs.get_secret.outputFiles[execution.id]) }}"
    kvType: JSON
    ttl: PT15S
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 10
      warningOnRetry: false

outputs:
  - id: secrets
    type: SECRET
    value: "{{ kv(execution.id) }}"

afterExecution:
  - id: delete_storage_file
    type: io.kestra.plugin.core.storage.Delete
    uri: "{{ outputs.get_secret.outputFiles[execution.id] }}"
    retry:
      type: constant
      interval: PT30S
      maxAttempts: 10
      warningOnRetry: false

From my observations this happens when a Subflow output of type SECRET is used in another flow. If you remove the get_secret subflow task from the flow_secret_output_issue and run it or if you run get_secret directly, it succeeds!

Environment

• Kestra Version: v1.3.2