Repository: kestra-io/kestra (java)

Issue:
Request to make `timeout` Flow.Pause Property Dynamic
### Feature description

Is it possible if Flow.Pause timeout property could be Dynamic? https://kestra.io/plugins/core/tasks/flows/io.kestra.plugin.core.flow.pause#timeout

My reasoning is I have Pause in a Subflow, and some flows will have shorter timeouts than others. Some flows will need to always be reviewed, while some will be fine without review and can have auto-approve/timeout hit.

Also worth noting is that we use Pause in a subflow, because we built slack/resuming integration that is reusable among many workflows.

This would be super helpful, but if not possible, we can make adjustments so Pause is in every flow instead of called by subflow. Thought it was worth requesting if possible to do.

Thank you 🙏 

Pull request description:
Feat/dynamic pause
Fixes #5338

Allow passing the Pause delay or timeout property as a Pebble expression:

```yaml
id: pause
namespace: company.team

inputs:
  - id: timeout
    type: DURATION
    defaults: PT10S

tasks:
  - id: pause
    type: io.kestra.plugin.core.flow.Pause
    delay: "{{inputs.timeout}}"
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: Hello World! 🚀
```

@anna-geller in fact it allows passing any task's timeout property as a Pebble expression, I had to do it ;)

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/property/Property.java b/core/src/main/java/io/kestra/core/models/property/Property.java
index 44050f78a..48c777723 100644
--- a/core/src/main/java/io/kestra/core/models/property/Property.java
+++ b/core/src/main/java/io/kestra/core/models/property/Property.java
@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.io.Serial;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Define a plugin properties that will be rendered and converted to a target type at use time.
@@ -35,7 +36,7 @@ import java.util.Map;
 public class Property<T> {
     // By default, durations are stored as numbers.
     // We cannot change that globally, as in JDBC/Elastic 'execution.state.duration' must be a number to be able to aggregate them.
-    // So we only change it here.
+    // So we only change it here to be used for Property.of().
     private static final ObjectMapper MAPPER = JacksonMapper.ofJson()
         .copy()
         .configure(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS, false);
@@ -91,6 +92,7 @@ public class Property<T> {
     public static <T> T as(Property<T> property, RunContext runContext, Class<T> clazz) throws IllegalVariableEvaluationException {
         if (property.value == null) {
             String rendered =  runContext.render(property.expression);
+            // special case for duration as they should be serialized as double but are not always
             property.value = MAPPER.convertValue(rendered, clazz);
         }
 
@@ -331,6 +333,18 @@ public class Property<T> {
         return value != null ? value.toString() : expression;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        Property<?> property = (Property<?>) o;
+        return Objects.equals(expression, property.expression);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(expression);
+    }
+
     // used only by the serializer
     String getExpression() {
         return this.expression;
diff --git a/core/src/main/java/io/kestra/core/models/tasks/Task.java b/core/src/main/java/io/kestra/core/models/tasks/Task.java
index 69b0fb4c2..3ee2f993b 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/Task.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/Task.java
@@ -5,6 +5,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.core.flow.WorkingDirectory;
@@ -35,7 +36,7 @@ abstract public class Task implements TaskInterface {
     @Valid
     protected AbstractRetry retry;
 
-    protected Duration timeout;
+    protected Property<Duration> timeout;
 
     @Builder.Default
     protected Boolean disabled = false;
diff --git a/core/src/main/java/io/kestra/core/plugins/RegisteredPlugin.java b/core/src/main/java/io/kestra/core/plugins/RegisteredPlugin.java
index 3f3b71686..dfb08ad37 100644
--- a/core/src/main/java/io/kestra/core/plugins/RegisteredPlugin.java
+++ b/core/src/main/java/io/kestra/core/plugins/RegisteredPlugin.java
@@ -111,7 +111,15 @@ public class RegisteredPlugin {
             return DataFilter.class;
         }
 
-        if(this.getAliases().containsKey(cls.toLowerCase())) {
+        if (this.getAppBlocks().stream().anyMatch(r -> r.getName().equals(cls))) {
+            return AppBlockInterface.class;
+        }
+
+        if (this.getApps().stream().anyMatch(r -> r.getName().equals(cls))) {
+            return AppPluginInterface.class;
+        }
+
+        if (this.getAliases().containsKey(cls.toLowerCase())) {
             // This is a quick-win, but it may trigger an infinite loop ... or not ...
             return baseClass(this.getAliases().get(cls.toLowerCase()).getValue().getName());
         }
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
index 45ee06f55..336c0c785 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
@@ -19,6 +19,8 @@ import java.time.Instant;
 import java.time.ZonedDateTime;
 import java.util.*;
 
+import static io.kestra.core.utils.Rethrow.throwConsumer;
+
 @Slf4j
 public final class ExecutableUtils {
 
@@ -88,7 +90,7 @@ public final class ExecutableUtils {
 
         List<Label> newLabels = inheritLabels ? new ArrayList<>(currentExecution.getLabels()) : new ArrayList<>(systemLabels(currentExecution));
         if (labels != null) {
-            newLabels.addAll(labels);
+            labels.forEach(throwConsumer(label -> newLabels.add(new Label(runContext.render(label.key()), runContext.render(label.value())))));
         }
 
         Map<String, Object> variables = ImmutableMap.of(
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
index 047e0d9cc..80200c185 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -28,6 +28,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.event.Level;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
 import java.util.stream.Collectors;
@@ -618,13 +619,18 @@ public class ExecutorService {
 
                 if (task instanceof Pause pauseTask) {
                     if (pauseTask.getDelay() != null || pauseTask.getTimeout() != null) {
-                        return ExecutionDelay.builder()
-                            .taskRunId(workerTaskResult.getTaskRun().getId())
-                            .executionId(executor.getExecution().getId())
-                            .date(workerTaskResult.getTaskRun().getState().maxDate().plus(pauseTask.getDelay() != null ? pauseTask.getDelay() : pauseTask.getTimeout()))
-                            .state(pauseTask.getDelay() != null ? State.Type.RUNNING : State.Type.FAILED)
-                            .delayType(ExecutionDelay.DelayType.RESUME_FLOW)
-                            .build();
+                        RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
+                        Duration delay = runContext.render(pauseTask.getDelay()).as(Duration.class).orElse(null);
+                        Duration timeout = runContext.render(pauseTask.getTimeout()).as(Duration.class).orElse(null);
+                        if (delay != null || timeout != null) { // rendering can lead to null, so we must re-check here
+                            return ExecutionDelay.builder()
+                                .taskRunId(workerTaskResult.getTaskRun().getId())
+                                .executionId(executor.getExecution().getId())
+                                .date(workerTaskResult.getTaskRun().getState().maxDate().plus(delay != null ? delay : timeout))
+                                .state(delay != null ? State.Type.RUNNING : State.Type.FAILED)
+                                .delayType(ExecutionDelay.DelayType.RESUME_FLOW)
+                                .build();
+                        }
                     }
                 }
 
diff --git a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
index 90ce27351..57f51aafd 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
@@ -35,8 +35,6 @@ import jakarta.validation.ConstraintViolationException;
 import jakarta.validation.constraints.NotNull;
 
 import org.reactivestreams.Publisher;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
@@ -73,10 +71,10 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
  */
 @Singleton
 public class FlowInputOutput {
-    private static final Logger log = LoggerFactory.getLogger(FlowInputOutput.class);
+    private static final Pattern URI_PATTERN = Pattern.compile("^[a-z]+:\\/\\/(?:www\\.)?[-a-zA-Z0-9@:%._\\+~#=]{1,256}\\.[a-zA-Z0-9()]{1,6}\\b(?:[-a-zA-Z0-9()@:%_\\+.~#?&\\/=]*)$");
+    private static final ObjectMapper YAML_MAPPER = JacksonMapper.ofYaml();
+    private static final ObjectMapper JSON_MAPPER = JacksonMapper.ofJson();
 
-    public static final Pattern URI_PATTERN = Pattern.compile("^[a-z]+:\\/\\/(?:www\\.)?[-a-zA-Z0-9@:%._\\+~#=]{1,256}\\.[a-zA-Z0-9()]{1,6}\\b(?:[-a-zA-Z0-9()@:%_\\+.~#?&\\/=]*)$");
-    public static final ObjectMapper YAML_MAPPER = JacksonMapper.ofYaml();
     private final StorageInterface storageInterface;
     private final Optional<String> secretKey;
     private final RunContextFactory runContextFactory;
@@ -358,7 +356,6 @@ public class FlowInputOutput {
         if (flow.getOutputs() == null) {
             return ImmutableMap.of();
         }
-        final ObjectMapper mapper = new ObjectMapper();
         Map<String, Object> results = flow
             .getOutputs()
             .stream()
@@ -366,8 +363,8 @@ public class FlowInputOutput {
                 final HashMap<String, Object> current;
                 final Object currentValue;
                 try {
-                    current = in == null ? null : mapper.readValue(
-                        mapper.writeValueAsString(in.get(output.getId())), new TypeReference<>() {});
+                    current = in == null ? null : JSON_MAPPER.readValue(
+                        JSON_MAPPER.writeValueAsString(in.get(output.getId())), new TypeReference<>() {});
                 } catch (JsonProcessingException e) {
                     throw new RuntimeException(e);
                 }
diff --git a/core/src/main/java/io/kestra/core/runners/WorkerTaskCallable.java b/core/src/main/java/io/kestra/core/runners/WorkerTaskCallable.java
index dabbc890c..5ec1dd941 100644
--- a/core/src/main/java/io/kestra/core/runners/WorkerTaskCallable.java
+++ b/core/src/main/java/io/kestra/core/runners/WorkerTaskCallable.java
@@ -53,7 +53,7 @@ public class WorkerTaskCallable extends AbstractWorkerCallable {
 
     @Override
     public State.Type doCall() throws Exception {
-        final Duration workerTaskTimeout = workerTask.getTask().getTimeout();
+        final Duration workerTaskTimeout = workerTask.getRunContext().render(workerTask.getTask().getTimeout()).as(Duration.class).orElse(null);
 
         try {
             if (workerTaskTimeout != null) {
diff --git a/core/src/main/java/io/kestra/core/serializers/DurationDeserializer.java b/core/src/main/java/io/kestra/core/serializers/DurationDeserializer.java
new file mode 100644
index 000000000..cd46e3c27
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/serializers/DurationDeserializer.java
@@ -0,0 +1,77 @@
+package io.kestra.core.serializers;
+
+import com.fasterxml.jackson.core.*;
+import com.fasterxml.jackson.core.io.NumberInput;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.datatype.jsr310.DecimalUtils;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.DateTimeException;
+import java.time.Duration;
+
+public class DurationDeserializer extends com.fasterxml.jackson.datatype.jsr310.deser.DurationDeserializer {
+
+    // durations can be a string with a number which is not taken into account as it should not happen
+    // we specialize the Duration deserialization from string to support that
+    @Override
+    protected Duration _fromString(JsonParser parser, DeserializationContext ctxt, String value0) throws IOException {
+        String value = value0.trim();
+        if (value.isEmpty()) {
+            // 22-Oct-2020, tatu: not sure if we should pass original (to distinguish
+            //   b/w empty and blank); for now don't which will allow blanks to be
+            //   handled like "regular" empty (same as pre-2.12)
+            return _fromEmptyString(parser, ctxt, value);
+        }
+        // 30-Sep-2020: Should allow use of "Timestamp as String" for
+        //     some textual formats
+        if (ctxt.isEnabled(StreamReadCapability.UNTYPED_SCALARS)
+            && _isValidTimestampString(value)) {
+            return _fromTimestamp(ctxt, NumberInput.parseLong(value));
+        }
+
+        // These are the only lines we changed from the default impl: we check for a float as string and parse it
+        if (_isFloat(value)) {
+            double d = Double.parseDouble(value);
+            BigDecimal bigDecimal = BigDecimal.valueOf(d);
+            return DecimalUtils.extractSecondsAndNanos(bigDecimal, Duration::ofSeconds);
+        }
+
+        try {
+            return Duration.parse(value);
+        } catch (DateTimeException e) {
+            return _handleDateTimeException(ctxt, e, value);
+        }
+    }
+
+    // this method is inspired by _isIntNumber but allow the decimal separator '.'
+    private boolean _isFloat(String text) {
+        final int len = text.length();
+        if (len > 0) {
+            char c = text.charAt(0);
+            // skip leading sign (plus not allowed for strict JSON numbers but...)
+            int i;
+
+            if (c == '-' || c == '+') {
+                if (len == 1) {
+                    return false;
+                }
+                i = 1;
+            } else {
+                i = 0;
+            }
+            // We will allow leading
+            for (; i < len; ++i) {
+                int ch = text.charAt(i);
+                if (ch == '.') {
+                    continue;
+                }
+                if (ch > '9' || ch < '0') {
+                    return false;
+                }
+            }
+            return true;
+        }
+        return false;
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/serializers/JacksonMapper.java b/core/src/main/java/io/kestra/core/serializers/JacksonMapper.java
index abcb9528f..cbbf93e76 100644
--- a/core/src/main/java/io/kestra/core/serializers/JacksonMapper.java
+++ b/core/src/main/java/io/kestra/core/serializers/JacksonMapper.java
@@ -10,6 +10,7 @@ import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.dataformat.ion.IonObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
@@ -29,6 +30,7 @@ import org.apache.commons.lang3.tuple.Pair;
 import org.yaml.snakeyaml.LoaderOptions;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.time.ZoneId;
 import java.util.List;
 import java.util.Map;
@@ -119,6 +121,9 @@ public final class JacksonMapper {
     }
 
     private static ObjectMapper configure(ObjectMapper mapper) {
+        SimpleModule durationDeserialization = new SimpleModule();
+        durationDeserialization.addDeserializer(Duration.class, new DurationDeserializer());
+
         return mapper
             .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
             .setSerializationInclusion(JsonInclude.Include.NON_NULL)
@@ -128,6 +133,7 @@ public final class JacksonMapper {
             .registerModules(new GuavaModule())
             .registerModule(new PluginModule())
             .registerModule(new RunContextModule())
+            .registerModule(durationDeserialization)
             .setTimeZone(TimeZone.getDefault());
     }
 
diff --git a/core/src/main/java/io/kestra/plugin/core/flow/Pause.java b/core/src/main/java/io/kestra/plugin/core/flow/Pause.java
index fff3e7a17..9b9fd0c8f 100644
--- a/core/src/main/java/io/kestra/plugin/core/flow/Pause.java
+++ b/core/src/main/java/io/kestra/plugin/core/flow/Pause.java
@@ -13,6 +13,7 @@ import io.kestra.core.models.hierarchies.AbstractGraph;
 import io.kestra.core.models.hierarchies.GraphCluster;
 import io.kestra.core.models.hierarchies.GraphTask;
 import io.kestra.core.models.hierarchies.RelationType;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
@@ -135,15 +136,13 @@ public class Pause extends Task implements FlowableTask<Pause.Output> {
         title = "Duration of the pause — useful if you want to pause the execution for a fixed amount of time.",
         description = "The delay is a string in the [ISO 8601 Duration](https://en.wikipedia.org/wiki/ISO_8601#Durations) format, e.g. `PT1H` for 1 hour, `PT30M` for 30 minutes, `PT10S` for 10 seconds, `P1D` for 1 day, etc. If no delay and no timeout are configured, the execution will never end until it's manually resumed from the UI or API."
     )
-    @PluginProperty
-    private Duration delay;
+    private Property<Duration> delay;
 
     @Schema(
         title = "Timeout of the pause — useful to avoid never-ending workflows in a human-in-the-loop scenario. For example, if you want to pause the execution until a human validates some data generated in a previous task, you can set a timeout of e.g. 24 hours. If no manual approval happens within 24 hours, the execution will automatically resume without a prior data validation.",
         description = "If no delay and no timeout are configured, the execution will never end until it's manually resumed from the UI or API."
     )
-    @PluginProperty
-    private Duration timeout;
+    private Property<Duration> timeout;
 
     @Valid
     @Schema(
diff --git a/core/src/main/resources/docs/task.peb b/core/src/main/resources/docs/task.peb
index ef1aff4bd..4e73a1bee 100644
--- a/core/src/main/resources/docs/task.peb
+++ b/core/src/main/resources/docs/task.peb
@@ -79,7 +79,7 @@ type: "{{ cls }}"
 {% endif %}
 
 
-{% if outputs != null -%}
+{% if outputs != null and outputs|length > 0 -%}
 ## Outputs
     {%- for entry in outputs %}
 ### `{{ entry.key }}`
@@ -90,7 +90,7 @@ type: "{{ cls }}"
 {% endif %}
 
 
-{% if defs != null -%}
+{% if defs != null and defs|length > 0 -%}
 ## Definitions
     {%- for entry in defs %}
 ### `{{ entry.key }}`
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
index 7a44ee4a6..c588cea80 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
@@ -352,6 +352,9 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             throw new UnsupportedOperationException();
         }
 
+        ZonedDateTime finalStartDate = startDate == null ? ZonedDateTime.now().minusDays(30) : startDate;
+        ZonedDateTime finalEndDate = endDate == null ? ZonedDateTime.now() : endDate;
+
         Results results = dailyStatisticsQueryForAllTenants(
             List.of(
                 STATE_CURRENT_FIELD
@@ -360,8 +363,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             namespace,
             flowId,
             null,
-            startDate,
-            endDate,
+            finalStartDate,
+            finalEndDate,
             groupBy,
             null
         );
@@ -370,8 +373,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             results.resultsOrRows()
                 .getFirst()
                 .result(),
-            startDate,
-            endDate,
+            finalStartDate,
+            finalEndDate,
             groupBy
         );
     }
@@ -393,6 +396,9 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             throw new UnsupportedOperationException();
         }
 
+        ZonedDateTime finalStartDate = startDate == null ? ZonedDateTime.now().minusDays(30) : startDate;
+        ZonedDateTime finalEndDate = endDate == null ? ZonedDateTime.now() : endDate;
+
         Results results = dailyStatisticsQuery(
             List.of(
                 STATE_CURRENT_FIELD
@@ -403,8 +409,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             namespace,
             flowId,
             null,
-            startDate,
-            endDate,
+            finalStartDate,
+            finalEndDate,
             groupBy,
             states
         );
@@ -413,8 +419,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             results.resultsOrRows()
                 .getFirst()
                 .result(),
-            startDate,
-            endDate,
+            finalStartDate,
+            finalEndDate,
             groupBy
         );
     }
@@ -453,8 +459,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
         @Nullable String namespace,
         @Nullable String flowId,
         List<FlowFilter> flows,
-        @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate,
+        ZonedDateTime startDate,
+        ZonedDateTime endDate,
         @Nullable DateUtils.GroupType groupBy,
         @Nullable List<State.Type> state
     ) {
@@ -481,8 +487,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
         @Nullable String namespace,
         @Nullable String flowId,
         List<FlowFilter> flows,
-        @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate,
+        ZonedDateTime startDate,
+        ZonedDateTime endDate,
         @Nullable DateUtils.GroupType groupBy,
         @Nullable List<State.Type> state
     ) {
@@ -509,15 +515,12 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
         @Nullable String namespace,
         @Nullable String flowId,
         List<FlowFilter> flows,
-        @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate,
+        ZonedDateTime startDate,
+        ZonedDateTime endDate,
         @Nullable DateUtils.GroupType groupBy,
         @Nullable List<State.Type> state
     ) {
-        ZonedDateTime finalStartDate = startDate == null ? ZonedDateTime.now().minusDays(30) : startDate;
-        ZonedDateTime finalEndDate = endDate == null ? ZonedDateTime.now() : endDate;
-
-        List<Field<?>> dateFields = new ArrayList<>(groupByFields(Duration.between(finalStartDate, finalEndDate), "start_date", groupBy));
+        List<Field<?>> dateFields = new ArrayList<>(groupByFields(Duration.between(startDate, endDate), "start_date", groupBy));
         List<Field<?>> selectFields = new ArrayList<>(fields);
         selectFields.addAll(List.of(
             DSL.count().as("count"),
@@ -525,7 +528,7 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             DSL.max(field("state_duration", Long.class)).as("duration_max"),
             DSL.sum(field("state_duration", Long.class)).as("duration_sum")
         ));
-        selectFields.addAll(groupByFields(Duration.between(finalStartDate, finalEndDate), "start_date", groupBy, true));
+        selectFields.addAll(groupByFields(Duration.between(startDate, endDate), "start_date", groupBy, true));
 
         return jdbcRepository
             .getDslContextWrapper()
@@ -536,8 +539,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                     .select(selectFields)
                     .from(this.jdbcRepository.getTable())
                     .where(defaultFilter)
-                    .and(START_DATE_FIELD.greaterOrEqual(finalStartDate.toOffsetDateTime()))
-                    .and(START_DATE_FIELD.lessOrEqual(finalEndDate.toOffsetDateTime()));
+                    .and(START_DATE_FIELD.greaterOrEqual(startDate.toOffsetDateTime()))
+                    .and(START_DATE_FIELD.lessOrEqual(endDate.toOffsetDateTime()));
 
                 select = filteringQuery(select, scope, namespace, flowId, flows, query, null, null, null);
 
@@ -680,6 +683,9 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             fields.add(field("flow_id", String.class));
         }
 
+        ZonedDateTime finalStartDate = startDate == null ? ZonedDateTime.now().minusDays(30) : startDate;
+        ZonedDateTime finalEndDate = endDate == null ? ZonedDateTime.now() : endDate;
+
         Results results = dailyStatisticsQuery(
             fields,
             query,
@@ -688,8 +694,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
             namespace,
             flowId,
             flows,
-            startDate,
-            endDate,
+            finalStartDate,
+            finalEndDate,
             null,
             null
         );
@@ -709,8 +715,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                             "*",
                             dailyStatisticsQueryMapRecord(
                                 e.getValue(),
-                                startDate,
-                                endDate,
+                                finalStartDate,
+                                finalEndDate,
                                 null
                             )
                         )
@@ -725,8 +731,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                                 f.getKey(),
                                 dailyStatisticsQueryMapRecord(
                                     f.getValue(),
-                                    startDate,
-                                    endDate,
+                                    finalStartDate,
+                                    finalEndDate,
                                     null
                                 )
                             ))
@@ -843,13 +849,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                     .from(this.jdbcRepository.getTable())
                     .where(this.defaultFilter(tenantId));
 
-                if (startDate != null) {
-                    select = select.and(START_DATE_FIELD.greaterOrEqual(finalStartDate.toOffsetDateTime()));
-                }
-
-                if (endDate != null) {
-                    select = select.and(field("end_date").lessOrEqual(finalEndDate.toOffsetDateTime()));
-                }
+                select = select.and(START_DATE_FIELD.greaterOrEqual(finalStartDate.toOffsetDateTime()));
+                select = select.and(field("end_date").lessOrEqual(finalEndDate.toOffsetDateTime()));
 
                 if (states != null) {
                     select = select.and(this.statesFilter(states));
diff --git a/script/src/main/java/io/kestra/plugin/scripts/exec/AbstractExecScript.java b/script/src/main/java/io/kestra/plugin/scripts/exec/AbstractExecScript.java
index 6cc448e8a..5d42adcad 100644
--- a/script/src/main/java/io/kestra/plugin/scripts/exec/AbstractExecScript.java
+++ b/script/src/main/java/io/kestra/plugin/scripts/exec/AbstractExecScript.java
@@ -20,6 +20,7 @@ import lombok.*;
 import lombok.experimental.SuperBuilder;
 import org.apache.commons.lang3.SystemUtils;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -161,7 +162,7 @@ public abstract class AbstractExecScript extends Task implements RunnableTask<Sc
             .withInputFiles(this.getInputFiles())
             .withOutputFiles(this.getOutputFiles())
             .withEnableOutputDirectory(this.getOutputDirectory())
-            .withTimeout(this.getTimeout())
+            .withTimeout(runContext.render(this.getTimeout()).as(Duration.class).orElse(null))
             .withTargetOS(this.getTargetOS());
     }
 
diff --git a/ui/package.json b/ui/package.json
index 7e8a09e01..d2d999716 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -15,7 +15,7 @@
   },
   "dependencies": {
     "@js-joda/core": "^5.6.3",
-    "@kestra-io/ui-libs": "^0.0.69",
+    "@kestra-io/ui-libs": "^0.0.72",
     "@vue-flow/background": "^1.3.2",
     "@vue-flow/controls": "^1.1.2",
     "@vue-flow/core": "^1.41.5",
diff --git a/ui/src/components/plugins/PluginDocumentation.vue b/ui/src/components/plugins/PluginDocumentation.vue
index eaeaaa7c5..06aa97cb2 100644
--- a/ui/src/components/plugins/PluginDocumentation.vue
+++ b/ui/src/components/plugins/PluginDocumentation.vue
@@ -13,10 +13,16 @@
     export default {
         name: "PluginDocumentation",
         components: {Markdown},
+        props: {
+            overrideIntro: {
+                type: String,
+                default: null
+            }
+        },
         computed: {
             ...mapState("plugin", ["editorPlugin"]),
             introContent () {
-                return intro
+                return this.overrideIntro ?? intro
             }
         },
         created() {


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/WorkerTest.java b/core/src/test/java/io/kestra/core/runners/WorkerTest.java
index 7bb700c7b..900ca39fa 100644
--- a/core/src/test/java/io/kestra/core/runners/WorkerTest.java
+++ b/core/src/test/java/io/kestra/core/runners/WorkerTest.java
@@ -5,6 +5,7 @@ import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
@@ -97,7 +98,7 @@ class WorkerTest {
 
         Pause pause = Pause.builder()
             .type(Pause.class.getName())
-            .delay(Duration.ofSeconds(1))
+            .delay(Property.of(Duration.ofSeconds(1)))
             .id("unit-test")
             .build();
 
diff --git a/core/src/test/java/io/kestra/core/serializers/YamlParserTest.java b/core/src/test/java/io/kestra/core/serializers/YamlParserTest.java
index 9071d99c5..e11afde5e 100644
--- a/core/src/test/java/io/kestra/core/serializers/YamlParserTest.java
+++ b/core/src/test/java/io/kestra/core/serializers/YamlParserTest.java
@@ -6,6 +6,7 @@ import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.Type;
 import io.kestra.core.models.flows.input.StringInput;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.Constant;
 import io.kestra.core.models.validations.ModelValidator;
@@ -32,7 +33,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @KestraTest
 class YamlParserTest {
-    private static ObjectMapper mapper = JacksonMapper.ofJson();
+    private static final ObjectMapper MAPPER = JacksonMapper.ofJson();
 
     @Inject
     private YamlParser yamlParser;
@@ -49,7 +50,7 @@ class YamlParserTest {
 
         // third with all optionals
         Task optionals = flow.getTasks().get(2);
-        assertThat(optionals.getTimeout(), is(Duration.ofMinutes(60)));
+        assertThat(optionals.getTimeout(), is(Property.builder().expression("PT60M").build()));
         assertThat(optionals.getRetry().getType(), is("constant"));
         assertThat(optionals.getRetry().getMaxAttempt(), is(5));
         assertThat(((Constant) optionals.getRetry()).getInterval().getSeconds(), is(900L));
@@ -65,7 +66,7 @@ class YamlParserTest {
 
         // third with all optionals
         Task optionals = flow.getTasks().get(2);
-        assertThat(optionals.getTimeout(), is(Duration.ofMinutes(60)));
+        assertThat(optionals.getTimeout(), is(Property.builder().expression("PT60M").build()));
         assertThat(optionals.getRetry().getType(), is("constant"));
         assertThat(optionals.getRetry().getMaxAttempt(), is(5));
         assertThat(((Constant) optionals.getRetry()).getInterval().getSeconds(), is(900L));
@@ -166,7 +167,7 @@ class YamlParserTest {
     void serialization() throws IOException {
         Flow flow = this.parse("flows/valids/minimal.yaml");
 
-        String s = mapper.writeValueAsString(flow);
+        String s = MAPPER.writeValueAsString(flow);
         assertThat(s, is("{\"id\":\"minimal\",\"namespace\":\"io.kestra.tests\",\"revision\":2,\"disabled\":false,\"deleted\":false,\"labels\":[{\"key\":\"system.readOnly\",\"value\":\"true\"}],\"tasks\":[{\"id\":\"date\",\"type\":\"io.kestra.plugin.core.debug.Return\",\"format\":\"{{taskrun.startDate}}\"}]}"));
     }
 
@@ -174,7 +175,7 @@ class YamlParserTest {
     void noDefault() throws IOException {
         Flow flow = this.parse("flows/valids/parallel.yaml");
 
-        String s = mapper.writeValueAsString(flow);
+        String s = MAPPER.writeValueAsString(flow);
         assertThat(s, not(containsString("\"-c\"")));
         assertThat(s, containsString("\"deleted\":false"));
     }
diff --git a/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java b/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java
index 898c7bb79..055193351 100644
--- a/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java
+++ b/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.validations;
 
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.models.validations.ModelValidator;
 import io.kestra.plugin.core.flow.Pause;
@@ -57,7 +58,7 @@ public class WorkingDirectoryTest {
                 List.of(Pause.builder()
                     .id("pause")
                     .type(Pause.class.getName())
-                    .delay(Duration.ofSeconds(1L))
+                    .delay(Property.of(Duration.ofSeconds(1L)))
                     .build()
                 )
             )
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
index 912d0b362..4442eeeb7 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
@@ -56,6 +56,11 @@ public class PauseTest extends AbstractMemoryRunnerTest {
         suite.runDelay(runnerUtils);
     }
 
+    @Disabled("This test is too flaky and it always pass in JDBC and Kafka")
+    void delayFromInput() throws Exception {
+        suite.runDelayFromInput(runnerUtils);
+    }
+
     @Disabled("This test is too flaky and it always pass in JDBC and Kafka")
     void parallelDelay() throws Exception {
         suite.runParallelDelay(runnerUtils);
@@ -144,6 +149,24 @@ public class PauseTest extends AbstractMemoryRunnerTest {
             assertThat(execution.getTaskRunList(), hasSize(3));
         }
 
+        public void runDelayFromInput(RunnerUtils runnerUtils) throws Exception {
+            Execution execution = runnerUtils.runOneUntilPaused(null, "io.kestra.tests", "pause-delay-from-input", null, null, Duration.ofSeconds(30));
+            String executionId = execution.getId();
+
+            assertThat(execution.getState().getCurrent(), is(State.Type.PAUSED));
+            assertThat(execution.getTaskRunList(), hasSize(1));
+
+            execution = runnerUtils.awaitExecution(
+                e -> e.getId().equals(executionId) && e.getState().getCurrent() == State.Type.SUCCESS,
+                () -> {},
+                Duration.ofSeconds(5)
+            );
+
+            assertThat(execution.getTaskRunList().getFirst().getState().getHistories().stream().filter(history -> history.getState() == State.Type.PAUSED).count(), is(1L));
+            assertThat(execution.getTaskRunList().getFirst().getState().getHistories().stream().filter(history -> history.getState() == State.Type.RUNNING).count(), is(2L));
+            assertThat(execution.getTaskRunList(), hasSize(3));
+        }
+
         public void runParallelDelay(RunnerUtils runnerUtils) throws TimeoutException, QueueException {
             Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "each-parallel-pause", Duration.ofSeconds(30));
 
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/TimeoutTest.java b/core/src/test/java/io/kestra/plugin/core/flow/TimeoutTest.java
index eb2ed43cd..9ccc48195 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/TimeoutTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/TimeoutTest.java
@@ -4,6 +4,7 @@ import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -51,7 +52,7 @@ class TimeoutTest extends AbstractMemoryRunnerTest {
                 .id("test")
                 .type(Sleep.class.getName())
                 .duration(100000L)
-                .timeout(Duration.ofNanos(100000))
+                .timeout(Property.of(Duration.ofNanos(100000)))
                 .build()))
             .build();
 
diff --git a/core/src/test/resources/flows/valids/pause-delay-from-input.yaml b/core/src/test/resources/flows/valids/pause-delay-from-input.yaml
new file mode 100644
index 000000000..dff811fa1
--- /dev/null
+++ b/core/src/test/resources/flows/valids/pause-delay-from-input.yaml
@@ -0,0 +1,19 @@
+id: pause-delay-from-input
+namespace: io.kestra.tests
+
+inputs:
+  - id: delay
+    type: DURATION
+    defaults: PT1S
+
+tasks:
+  - id: pause
+    type: io.kestra.plugin.core.flow.Pause
+    delay: "{{inputs.delay}}"
+    tasks:
+      - id: subtask
+        type: io.kestra.plugin.core.log.Log
+        message: trigger 1 seconds pause"
+  - id: last
+    type: io.kestra.plugin.core.debug.Return
+    format: "{{task.id}} > {{taskrun.startDate}}"
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index d0c38081b..d1d6ac2c4 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -281,6 +281,11 @@ public abstract class JdbcRunnerTest {
         pauseTest.runDelay(runnerUtils);
     }
 
+    @Test
+    public void pauseRunDelayFromInput() throws Exception {
+        pauseTest.runDelayFromInput(runnerUtils);
+    }
+
     @Test
     public void pauseRunParallelDelay() throws Exception {
         pauseTest.runParallelDelay(runnerUtils);