Allow to re-execute a flow
### Feature description

It would be great to be able to re-run a job with the exact same parameters on the interface.

It would be useful for a flow with inputs (like a file).
Currently, we need to download the file and click "new execution" on a killed job for example.

We just had a bug with BigQuery not sending back any data, and had to kill our flow.
As they were curl called flows with a file  and a second parameter, the procedure to execute it again was a bit long and boring.

A button, like the "restart" button on failed execution would help greatly, but it would have to start the job from the very beginning instead of restarting at the failed step.

![image](https://user-images.githubusercontent.com/50616421/234864867-0e39d4ab-1866-4a69-bfb1-c4b47479f6a1.png)


core/src/main/java/io/kestra/core/services/ExecutionService.java ui/src/components/executions/Overview.vue ui/src/components/executions/Restart.vue ui/src/components/flows/FlowRun.vue ui/src/components/flows/TriggerFlow.vue ui/src/components/inputs/Editor.vue ui/src/components/logs/LogList.vue ui/src/translations.json ui/src/utils/submitTask.js
core/src/main/java/io/kestra/core/services/ExecutionService.java ui/src/components/executions/Overview.vue ui/src/components/executions/Restart.vue ui/src/components/flows/FlowRun.vue ui/src/components/flows/TriggerFlow.vue ui/src/components/inputs/Editor.vue ui/src/components/logs/LogList.vue ui/src/translations.json ui/src/utils/submitTask.js
core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
index 7ec700ad0c..f889369396 100644
--- a/core/src/main/java/io/kestra/core/services/ExecutionService.java
+++ b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -124,60 +124,59 @@ private Set<String> taskRunToRestart(Execution execution, Predicate<TaskRun> pre
         return finalTaskRunToRestart;
     }
 
-    public Execution replay(final Execution execution, String taskRunId, @Nullable Integer revision) throws Exception {
-        if (!(execution.getState().isTerminated() || !(execution.getState().isTerminated() ))) {
-            throw new IllegalStateException("Execution must be terminated to be restarted, " +
-                "current state is '" + execution.getState().getCurrent() + "' !"
-            );
-        }
-
-        final Flow flow = flowRepositoryInterface.findByExecution(execution);
-        GraphCluster graphCluster = GraphService.of(flow, execution);
-
-        Set<String> taskRunToRestart = this.taskRunToRestart(
-            execution,
-            taskRun -> taskRun.getId().equals(taskRunId)
-        );
-
-        Map<String, String> mappingTaskRunId = this.mapTaskRunId(execution, false);
+    public Execution replay(final Execution execution, @Nullable String taskRunId, @Nullable Integer revision) throws Exception {
         final String newExecutionId = IdUtils.create();
+        List<TaskRun> newTaskRuns = new ArrayList<>();
+        if(taskRunId != null){
+            final Flow flow = flowRepositoryInterface.findByExecution(execution);
 
-        List<TaskRun> newTaskRuns = execution
-            .getTaskRunList()
-            .stream()
-            .map(throwFunction(originalTaskRun -> this.mapTaskRun(
-                flow,
-                originalTaskRun,
-                mappingTaskRunId,
-                newExecutionId,
-                State.Type.RESTARTED,
-                taskRunToRestart.contains(originalTaskRun.getId()))
-            ))
-            .collect(Collectors.toList());
+            GraphCluster graphCluster = GraphService.of(flow, execution);
 
-        // remove all child for replay task id
-        Set<String> taskRunToRemove = GraphService.successors(graphCluster, List.of(taskRunId))
-            .stream()
-            .filter(task -> task instanceof AbstractGraphTask)
-            .map(task -> ((AbstractGraphTask) task))
-            .filter(task -> task.getTaskRun() != null)
-            .filter(task -> !task.getTaskRun().getId().equals(taskRunId))
-            .filter(task -> !taskRunToRestart.contains(task.getTaskRun().getId()))
-            .map(s -> mappingTaskRunId.get(s.getTaskRun().getId()))
-            .collect(Collectors.toSet());
+            Set<String> taskRunToRestart = this.taskRunToRestart(
+                execution,
+                taskRun -> taskRun.getId().equals(taskRunId)
+            );
 
-        taskRunToRemove
-            .forEach(r -> newTaskRuns.removeIf(taskRun -> taskRun.getId().equals(r)));
+            Map<String, String> mappingTaskRunId = this.mapTaskRunId(execution, false);
+
+            newTaskRuns.addAll(
+                execution.getTaskRunList()
+                .stream()
+                .map(throwFunction(originalTaskRun -> this.mapTaskRun(
+                    flow,
+                    originalTaskRun,
+                    mappingTaskRunId,
+                    newExecutionId,
+                    State.Type.RESTARTED,
+                    taskRunToRestart.contains(originalTaskRun.getId()))
+                ))
+                .collect(Collectors.toList())
+            );
 
-        // Worker task, we need to remove all child in order to be restarted
-        this.removeWorkerTask(flow, execution, taskRunToRestart, mappingTaskRunId)
-            .forEach(r -> newTaskRuns.removeIf(taskRun -> taskRun.getId().equals(r)));
+            // remove all child for replay task id
+            Set<String> taskRunToRemove = GraphService.successors(graphCluster, List.of(taskRunId))
+                .stream()
+                .filter(task -> task instanceof AbstractGraphTask)
+                .map(task -> ((AbstractGraphTask) task))
+                .filter(task -> task.getTaskRun() != null)
+                .filter(task -> !task.getTaskRun().getId().equals(taskRunId))
+                .filter(task -> !taskRunToRestart.contains(task.getTaskRun().getId()))
+                .map(s -> mappingTaskRunId.get(s.getTaskRun().getId()))
+                .collect(Collectors.toSet());
+
+            taskRunToRemove
+                .forEach(r -> newTaskRuns.removeIf(taskRun -> taskRun.getId().equals(r)));
+
+            // Worker task, we need to remove all child in order to be restarted
+            this.removeWorkerTask(flow, execution, taskRunToRestart, mappingTaskRunId)
+                .forEach(r -> newTaskRuns.removeIf(taskRun -> taskRun.getId().equals(r)));
+        }
 
         // Build and launch new execution
         Execution newExecution = execution.childExecution(
             newExecutionId,
             newTaskRuns,
-            execution.withState(State.Type.RESTARTED).getState()
+            taskRunId == null ? new State() : execution.withState(State.Type.RESTARTED).getState()
         );
 
         return revision != null ? newExecution.withFlowRevision(revision) : newExecution;
diff --git a/ui/src/components/executions/Overview.vue b/ui/src/components/executions/Overview.vue
index 039733569f..baf7f334fe 100644
--- a/ui/src/components/executions/Overview.vue
+++ b/ui/src/components/executions/Overview.vue
@@ -5,6 +5,7 @@
                 <crud type="CREATE" permission="EXECUTION" :detail="{executionId: execution.id}" />
             </el-col>
             <el-col :span="12" class="text-end">
+                <restart is-replay :execution="execution" @follow="forwardEvent('follow', $event)" />
                 <restart :execution="execution" @follow="forwardEvent('follow', $event)" />
                 <kill :execution="execution" />
                 <status :status="execution.state.current" />
diff --git a/ui/src/components/executions/Restart.vue b/ui/src/components/executions/Restart.vue
index 93c9c3046e..31f8ded728 100644
--- a/ui/src/components/executions/Restart.vue
+++ b/ui/src/components/executions/Restart.vue
@@ -1,15 +1,34 @@
 <template>
-    <component
-        :is="component"
-        :icon="!isReplay ? RestartIcon : PlayBoxMultiple"
-        @click="isOpen = !isOpen"
-        v-if="isReplay || enabled"
-        :disabled="!enabled"
-        :class="!isReplay ? 'restart me-1' : ''"
+    <el-tooltip
+        :persistent="false"
+        transition=""
+        :hide-after="0"
+        :content="tooltip"
+        raw-content
+        :placement="tooltipPosition"
     >
-        {{ $t(replayOrRestart) }}
-    </component>
-
+        <component
+            :is="component"
+            :icon="!isReplay ? RestartIcon : PlayBoxMultiple"
+            @click="isOpen = !isOpen"
+            v-if="component !== 'el-dropdown-item' && (isReplay || enabled)"
+            :disabled="!enabled"
+            :class="!isReplay ? 'restart me-1' : ''"
+        >
+            {{ $t(replayOrRestart) }}
+        </component>
+        <span v-else-if="component === 'el-dropdown-item' && (isReplay || enabled)">
+            <component
+                :is="component"
+                :icon="!isReplay ? RestartIcon : PlayBoxMultiple"
+                @click="isOpen = !isOpen"
+                :disabled="!enabled"
+                :class="!isReplay ? 'restart me-1' : ''"
+            >
+                {{ $t(replayOrRestart) }}
+            </component>
+        </span>
+    </el-tooltip>
     <el-dialog v-if="enabled && isOpen" v-model="isOpen" destroy-on-close :append-to-body="true">
         <template #header>
             <h5>{{ $t("confirmation") }}</h5>
@@ -84,6 +103,10 @@
                 type: Number,
                 required: false,
                 default: undefined
+            },
+            tooltipPosition: {
+                type: String,
+                default: "bottom"
             }
         },
         emits: ["follow"],
@@ -163,7 +186,7 @@
                     return false;
                 }
 
-                if (this.isReplay && (this.taskRun.attempts !== undefined && this.taskRun.attempts.length - 1 !== this.attemptIndex)) {
+                if (this.isReplay && (this.taskRun?.attempts !== undefined && this.taskRun.attempts.length - 1 !== this.attemptIndex)) {
                     return false;
                 }
 
@@ -173,6 +196,13 @@
 
                 return (this.isReplay && !State.isRunning(this.execution.state.current)) ||
                     (!this.isReplay && this.execution.state.current === State.FAILED);
+            },
+            tooltip(){
+                if(this.isReplay){
+                    return this?.taskRun?.id ? this.$t("replay from task tooltip", {taskId: this.taskRun.taskId}) :  this.$t("replay from beginning tooltip");
+                }
+
+                return this.$t("restart tooltip", {state: this.execution.state.current})
             }
         },
         data() {
diff --git a/ui/src/components/flows/FlowRun.vue b/ui/src/components/flows/FlowRun.vue
index e2df8e7f8f..99001427eb 100644
--- a/ui/src/components/flows/FlowRun.vue
+++ b/ui/src/components/flows/FlowRun.vue
@@ -53,11 +53,15 @@
                 <div class="el-input el-input-file">
                     <div class="el-input__wrapper" v-if="input.type === 'FILE'">
                         <input
+                            :id="input.name+'-file'"
                             class="el-input__inner"
                             type="file"
                             @change="onFileChange(input, $event)"
                             autocomplete="off"
+                            :style="{display: typeof(this.inputs[input.name]) === 'string' && this.inputs[input.name].startsWith('kestra:///') ? 'none': ''}"
                         >
+                        <label v-if="typeof(this.inputs[input.name]) === 'string' && this.inputs[input.name].startsWith('kestra:///')"
+                               :for="input.name+'-file'">Kestra Internal Storage File</label>
                     </div>
                 </div>
                 <editor
@@ -71,16 +75,28 @@
 
                 <small v-if="input.description" class="text-muted">{{ input.description }}</small>
             </el-form-item>
-            <el-form-item class="submit">
-                <el-button :icon="Flash" class="flow-run-trigger-button" @click="onSubmit($refs.form)" type="primary" :disabled="flow.disabled">
-                    {{ $t('launch execution') }}
-                </el-button>
-            </el-form-item>
+            <div class="bottom-buttons">
+                <div class="left-align">
+                    <el-form-item>
+                        <el-button v-if="execution && execution.inputs" :icon="ContentCopy" @click="fillInputsFromExecution">
+                            {{ $t('prefill inputs') }}
+                        </el-button>
+                    </el-form-item>
+                </div>
+                <div class="right-align">
+                    <el-form-item class="submit">
+                        <el-button :icon="Flash" class="flow-run-trigger-button" @click="onSubmit($refs.form)" type="primary" :disabled="flow.disabled">
+                            {{ $t('launch execution') }}
+                        </el-button>
+                    </el-form-item>
+                </div>
+            </div>
         </el-form>
     </div>
 </template>
 
 <script setup>
+    import ContentCopy from "vue-material-design-icons/ContentCopy.vue";
     import Flash from "vue-material-design-icons/Flash.vue";
 </script>
 
@@ -98,21 +114,22 @@
                 default: true
             }
         },
-        emits: ["executionTrigger"],
         data() {
             return {
-                inputs: {},
+                inputs: {}
             };
         },
-        mounted() {
+        emits: ["executionTrigger"],
+        created() {
             for (const input of this.flow.inputs || []) {
                 this.inputs[input.name] = input.defaults;
 
-                if (input.type === "DATETIME" && input.defaults) {
-                    this.inputs[input.name] = new Date(input.defaults);
+                if(input.type === "BOOLEAN" && input.defaults){
+                    this.inputs[input.name] = (/true/i).test(input.defaults);
                 }
             }
-
+        },
+        mounted() {
             setTimeout(() => {
                 const input = this.$el && this.$el.querySelector && this.$el.querySelector("input")
                 if (input && !input.className.includes("mx-input")) {
@@ -135,8 +152,29 @@
         computed: {
             ...mapState("flow", ["flow"]),
             ...mapState("core", ["guidedProperties"]),
+            ...mapState("execution", ["execution"]),
         },
         methods: {
+            fillInputsFromExecution(){
+                const nonEmptyInputNames = Object.keys(this.execution.inputs);
+                this.inputs = Object.fromEntries(
+                    this.flow.inputs.filter(input => nonEmptyInputNames.includes(input.name))
+                        .map(input => {
+                            const inputName = input.name;
+                            const inputType = input.type;
+                            let inputValue = this.execution.inputs[inputName];
+                            if (inputType === 'DATE' || inputType === 'DATETIME') {
+                                inputValue = this.$moment(inputValue).toISOString()
+                            }else if (inputType === 'DURATION' || inputType === 'TIME') {
+                                inputValue = this.$moment().startOf("day").add(inputValue, "seconds").toString()
+                            }else if (inputType === 'JSON') {
+                                inputValue = JSON.stringify(inputValue).toString()
+                            }
+
+                            return [inputName, inputValue]
+                        })
+                );
+            },
             onSubmit(formRef) {
                 if (this.$tours["guidedTour"].isRunning.value) {
                     this.finishTour();
@@ -226,3 +264,26 @@
         }
     };
 </script>
+
+<style scoped lang="scss">
+.bottom-buttons {
+    margin-top: 36px;
+    display: flex;
+
+    > * {
+        flex: 1;
+
+        * {
+            margin: 0;
+        }
+    }
+
+    .left-align :deep(div) {
+        flex-direction: row
+    }
+
+    .right-align :deep(div) {
+        flex-direction: row-reverse;
+    }
+}
+</style>
\ No newline at end of file
diff --git a/ui/src/components/flows/TriggerFlow.vue b/ui/src/components/flows/TriggerFlow.vue
index 845c54b12e..5f65ab7369 100644
--- a/ui/src/components/flows/TriggerFlow.vue
+++ b/ui/src/components/flows/TriggerFlow.vue
@@ -4,7 +4,7 @@
             {{ $t('New execution') }}
         </el-button>
         <el-dialog v-if="isOpen" v-model="isOpen" destroy-on-close :append-to-body="true">
-            <template #title>
+            <template #header>
                 <span v-html="$t('execute the flow', {id: flowId})" />
             </template>
             <flow-run @execution-trigger="closeModal" :redirect="true" />
@@ -42,7 +42,7 @@
             type: {
                 type: String,
                 default: "primary"
-            },
+            }
         },
         data() {
             return {
diff --git a/ui/src/components/inputs/Editor.vue b/ui/src/components/inputs/Editor.vue
index 74687420c9..b09e8fc3d7 100644
--- a/ui/src/components/inputs/Editor.vue
+++ b/ui/src/components/inputs/Editor.vue
@@ -33,7 +33,7 @@
                     :original="original"
                     @change="onInput"
                     @editor-did-mount="editorDidMount"
-                    :language="lang"
+                    :language="lang ?? 'undefined'"
                     :schema-type="schemaType"
                     class="position-relative"
                 />
diff --git a/ui/src/components/logs/LogList.vue b/ui/src/components/logs/LogList.vue
index 113cd7dbce..48d4fbca91 100644
--- a/ui/src/components/logs/LogList.vue
+++ b/ui/src/components/logs/LogList.vue
@@ -68,7 +68,8 @@
                                             <restart
                                                 component="el-dropdown-item"
                                                 :key="`restart-${index}-${attempt.state.startDate}`"
-                                                :is-replay="true"
+                                                is-replay
+                                                tooltip-position="left"
                                                 :execution="execution"
                                                 :task-run="currentTaskRun"
                                                 :attempt-index="index"
diff --git a/ui/src/translations.json b/ui/src/translations.json
index 413e6f4654..250e79b145 100644
--- a/ui/src/translations.json
+++ b/ui/src/translations.json
@@ -101,13 +101,17 @@
     "start datetime": "Start datetime",
     "end datetime": "End datetime",
     "restart": "Restart",
+    "restart tooltip": "Restart the execution from the <code>{state}</code> task",
     "restarted": "Execution is restarted",
     "restart confirm": "Are you sure to restart execution <code>{id}</code>?",
     "restart change revision": "You can change the revision that will be used for the new execution.",
     "replay": "Replay",
+    "replay from task tooltip": "Create a similar execution starting from task <code>{taskId}</code>",
+    "replay from beginning tooltip": "Create a similar execution starting from the beginning",
     "replay latest revision": "Replay using latest revision",
     "replayed": "Execution is replayed",
     "replay confirm": "Are you sure to replay this execution <code>{id}</code> and create a new one?",
+    "prefill inputs": "Prefill",
     "current": "current",
     "change status": "Change status",
     "change status done": "Task status has been updated",
@@ -531,13 +535,17 @@
     "start datetime": "Date de départ",
     "end datetime": "Date de fin",
     "restart": "Relancer",
+    "restart tooltip": "Relancer l'exécution depuis la tâche dans l'état <code>{state}</code>",
     "restarted": "L'exécution est relancée",
     "restart confirm": "Êtes-vous sur de vouloir relancer <code>{id}</code> ?",
     "restart change revision": "Vous pouvez changer la révision qui sera utilisé pour la nouvelle execution relancée.",
     "replay": "Rejouer",
+    "replay from task tooltip": "Créer une nouvelle exécution similaire démarrant de la tâche <code>{taskId}</code>",
+    "replay from beginning tooltip": "Créer une nouvelle exécution similaire depuis le début",
     "replay latest revision": "Rejouer avec la dernière révision",
     "replayed": "Execution est rejouée",
     "replay confirm": "Êtes-vous sur de vouloir relancer l'exécution <code>{id}</code> et créer une nouvelle execution ?",
+    "prefill inputs": "Pré-remplir",
     "current": "actuel",
     "change status": "Changer le statut",
     "change status done": "Le statut de la tâche a était mis à jour",
diff --git a/ui/src/utils/submitTask.js b/ui/src/utils/submitTask.js
index ef42aeb57e..604512a71b 100644
--- a/ui/src/utils/submitTask.js
+++ b/ui/src/utils/submitTask.js
@@ -1,23 +1,29 @@
 export const executeTask = (submitor, flow, values, options) => {
     const formData = new FormData();
     for (let input of flow.inputs || []) {
-        if (values[input.name] !== undefined) {
+        const inputName = input.name;
+        const inputValue = values[inputName];
+        if (inputValue !== undefined) {
             if (input.type === "DATETIME") {
-                formData.append(input.name, values[input.name].toISOString());
+                formData.append(inputName, submitor.$moment(inputValue).toISOString());
             } else if (input.type === "DATE") {
-                formData.append(input.name, submitor.$moment(values[input.name]).format("YYYY-MM-DD"));
+                formData.append(inputName, submitor.$moment(inputValue).format("YYYY-MM-DD"));
             } else if (input.type === "TIME") {
-                formData.append(input.name, submitor.$moment(values[input.name]).format("hh:mm:ss"));
+                formData.append(inputName, submitor.$moment(inputValue).format("hh:mm:ss"));
             } else if (input.type === "DURATION") {
-                formData.append(input.name, submitor.$moment.duration(submitor.$moment(values[input.name]).format("hh:mm:ss")));
+                formData.append(inputName, submitor.$moment.duration(submitor.$moment(inputValue).format("hh:mm:ss")));
             } else if (input.type === "FILE") {
-                formData.append("files", values[input.name], input.name);
+                if(typeof(inputValue) === "string"){
+                    formData.append(inputName, inputValue);
+                }else {
+                    formData.append("files", inputValue, inputName);
+                }
             } else {
-                formData.append(input.name, values[input.name]);
+                formData.append(inputName, inputValue);
             }
         } else if (input.required) {
             submitor.$toast().error(
-                submitor.$t("invalid field", {name: input.name}),
+                submitor.$t("invalid field", {name: inputName}),
                 submitor.$t("form error")
             )
 

diff --git a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
index eb18f687b1..7c337770e5 100644
--- a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
+++ b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
@@ -121,6 +121,26 @@ void restartDynamic() throws Exception {
         assertThat(restart.getTaskRunList().get(0).getState().getHistories(), hasSize(4));
     }
 
+    @Test
+    void replayFromBeginning() throws Exception {
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "logs");
+        assertThat(execution.getTaskRunList(), hasSize(3));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+
+        Execution restart = executionService.replay(execution, null, null);
+
+        assertThat(restart.getId(), not(execution.getId()));
+        assertThat(restart.getNamespace(), is("io.kestra.tests"));
+        assertThat(restart.getFlowId(), is("logs"));
+
+        assertThat(restart.getState().getCurrent(), is(State.Type.CREATED));
+        assertThat(restart.getState().getHistories(), hasSize(1));
+        assertThat(restart.getState().getHistories().get(0).getDate(), not(is(execution.getState().getStartDate())));
+        assertThat(restart.getTaskRunList(), hasSize(0));
+
+        assertThat(restart.getId(), not(execution.getId()));
+    }
+
     @Test
     void replaySimple() throws Exception {
         Execution execution = runnerUtils.runOne("io.kestra.tests", "logs");
