@Override
    public Optional<SubflowExecutionResult> createSubflowExecutionResult(
        RunContext runContext,
        TaskRun taskRun,
        io.kestra.core.models.flows.Flow flow,
        Execution execution
    ) {
        // we only create a worker task result when the execution is terminated
        if (!taskRun.getState().isTerminated()) {
            return Optional.empty();
        }

        boolean isOutputsAllowed = runContext
            .<Boolean>pluginConfiguration(PLUGIN_FLOW_OUTPUTS_ENABLED)
            .orElse(true);

        final Output.OutputBuilder builder = Output.builder()
            .executionId(execution.getId())
            .state(execution.getState().getCurrent());

        final Map<String, Object> subflowOutputs = Optional
            .ofNullable(flow.getOutputs())
            .map(outputs -> outputs
                .stream()
                .collect(Collectors.toMap(
                    io.kestra.core.models.flows.Output::getId,
                    io.kestra.core.models.flows.Output::getValue)
                )
            )
            .orElseGet(() -> isOutputsAllowed ? this.getOutputs() : null);

        if (subflowOutputs != null) {
            try {
                Map<String, Object> outputs = runContext.render(subflowOutputs);
                FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class); // this is hacking
                if (flow.getOutputs() != null && flowInputOutput != null) {
                    outputs = flowInputOutput.typedOutputs(flow, execution, outputs);
                }
                builder.outputs(outputs);
            } catch (Exception e) {
                runContext.logger().warn("Failed to extract outputs with the error: '{}'", e.getLocalizedMessage(), e);
                var state = this.isAllowFailure() ? State.Type.WARNING : State.Type.FAILED;
                taskRun = taskRun
                    .withState(state)
                    .withAttempts(Collections.singletonList(TaskRunAttempt.builder().state(new State().withState(state)).build()))
                    .withOutputs(builder.build().toMap());

                return Optional.of(SubflowExecutionResult.builder()
                    .executionId(execution.getId())
                    .state(State.Type.FAILED)
                    .parentTaskRun(taskRun)
                    .build());
            }
        }

        taskRun = taskRun.withOutputs(builder.build().toMap());

        State.Type finalState = ExecutableUtils.guessState(execution, this.transmitFailed, this.isAllowFailure());
        if (taskRun.getState().getCurrent() != finalState) {
            taskRun = taskRun.withState(finalState);
        }

        return Optional.of(ExecutableUtils.subflowExecutionResult(taskRun, execution));
    }
@Test
    void shouldNotReturnResultForExecutionNotTerminated() {
        TaskRun taskRun = TaskRun
            .builder()
            .state(State.of(State.Type.CREATED, Collections.emptyList()))
            .build();

        Optional<SubflowExecutionResult> result = new Subflow().createSubflowExecutionResult(
            runContext,
            taskRun,
            Flow.builder().build(),
            Execution.builder().build()
        );

        assertThat(result, is(Optional.empty()));
    }