Repository: kestra-io/kestra (java)

Issue:
Enhance multiple conditons with deadline
### Feature description

Currently, multiple condition for the flow trigger works with a sliding window of execution.
This is controlled by the `window` property with an additional `windowAdvance` property.

Sliding windows are a powerful concept but may be a bit hard to understand.

We can introduce a concept of deadline, which will always be on a daily basis.

For example, to launch a flow before 9am if flowA and flowB would be successfully executed we would be able to define the following flow trigger (using the new FlowFilters condition):

```yaml
id: myflow
namespace: company.team

triggers:
  - id: flow
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - id: flowFilters
        type: io.kestra.plugin.core.condition.FlowFilters
        deadline: 09:00:00
        upstreamFlows:
          - namespace: company.team
            flowId: low-a
            states: [SUCCESS, WARNING]
          - namespace: company.team
            flowId: flow-b
            states: [SUCCESS]

tasks:
  - id: only
    type: io.kestra.plugin.core.debug.Return
    format: "It works"
```

We can have the following semantics:
- `deadline`: the time at which we want the flow to be triggered.
- `latency`: optional, duration to wait for the conditions to be fulfill, by default 1mn (or 0 but it would be handled as 1mn in the executor so we have 1mn to trigger the flow at the deadline).
- `missedBehavior`: optional, what to do in case the deadline is missed, by default, nothing. Possible actions NOTHING, EXECUTE_SUCCESS, EXECUTE_FAIL.

Pull request description:
feat(core): MultipleConditon deadline
Fixes #5706

Patch:
diff --git a/cli/src/main/resources/application.yml b/cli/src/main/resources/application.yml
index 1576eb908..a8c14701b 100644
--- a/cli/src/main/resources/application.yml
+++ b/cli/src/main/resources/application.yml
@@ -191,5 +191,5 @@ kestra:
 
   hidden-labels:
     prefixes:
-      - system_
-      - internal_
+      - system.
+      - internal.
diff --git a/core/src/main/java/io/kestra/core/models/Label.java b/core/src/main/java/io/kestra/core/models/Label.java
index aa4be5153..d7664a325 100644
--- a/core/src/main/java/io/kestra/core/models/Label.java
+++ b/core/src/main/java/io/kestra/core/models/Label.java
@@ -3,7 +3,7 @@ package io.kestra.core.models;
 import jakarta.validation.constraints.NotNull;
 
 public record Label(@NotNull String key, @NotNull String value) {
-    public static final String SYSTEM_PREFIX = "system_";
+    public static final String SYSTEM_PREFIX = "system.";
 
     // system labels
     public static final String CORRELATION_ID = SYSTEM_PREFIX + "correlationId";
diff --git a/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleCondition.java b/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleCondition.java
index 50905aed2..d09e4b28f 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleCondition.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleCondition.java
@@ -1,16 +1,27 @@
 package io.kestra.core.models.triggers.multipleflows;
 
 import io.kestra.core.models.conditions.Condition;
+import lombok.Builder;
+import lombok.Getter;
 
 import java.time.Duration;
+import java.time.LocalTime;
 import java.util.Map;
 
 public interface MultipleCondition {
     String getId();
 
+    LatencySLA getLatencySLA();
+
     Duration getWindow();
 
     Duration getWindowAdvance();
 
     Map<String, Condition> getConditions();
+
+    @Getter
+    @Builder
+    class LatencySLA {
+        private LocalTime deadline;
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java b/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java
index 8811a208a..388d8bbe0 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionStorageInterface.java
@@ -15,25 +15,35 @@ public interface MultipleConditionStorageInterface {
     List<MultipleConditionWindow> expired(String tenantId);
 
     default MultipleConditionWindow getOrCreate(Flow flow, MultipleCondition multipleCondition) {
+
+        ZonedDateTime start;
+        ZonedDateTime end;
+
         ZonedDateTime now = ZonedDateTime.now()
             .withNano(0);
 
-        if (multipleCondition.getWindow().toDays() > 0) {
-            now = now.withHour(0);
-        }
+        if (multipleCondition.getLatencySLA() != null) {
+            // with latency, the start is always the start of the day
+            start = now.truncatedTo(ChronoUnit.DAYS);
+            end = start.plusSeconds(multipleCondition.getLatencySLA().getDeadline().toSecondOfDay());
+        } else {
+            if (multipleCondition.getWindow().toDays() > 0) {
+                now = now.withHour(0);
+            }
 
-        if (multipleCondition.getWindow().toHours() > 0) {
-            now = now.withMinute(0);
-        }
+            if (multipleCondition.getWindow().toHours() > 0) {
+                now = now.withMinute(0);
+            }
 
-        if (multipleCondition.getWindow().toMinutes() > 0) {
-            now = now.withSecond(0)
-                .withMinute(0)
-                .plusMinutes(multipleCondition.getWindow().toMinutes() * (now.getMinute() / multipleCondition.getWindow().toMinutes()));
-        }
+            if (multipleCondition.getWindow().toMinutes() > 0) {
+                now = now.withSecond(0)
+                    .withMinute(0)
+                    .plusMinutes(multipleCondition.getWindow().toMinutes() * (now.getMinute() / multipleCondition.getWindow().toMinutes()));
+            }
 
-        ZonedDateTime start = multipleCondition.getWindowAdvance() == null ? now : now.plus(multipleCondition.getWindowAdvance()).truncatedTo(ChronoUnit.MILLIS);
-        ZonedDateTime end = start.plus(multipleCondition.getWindow()).minus(Duration.ofMillis(1)).truncatedTo(ChronoUnit.MILLIS);
+            start = multipleCondition.getWindowAdvance() == null ? now : now.plus(multipleCondition.getWindowAdvance()).truncatedTo(ChronoUnit.MILLIS);
+            end = start.plus(multipleCondition.getWindow()).minus(Duration.ofMillis(1)).truncatedTo(ChronoUnit.MILLIS);
+        }
 
         return this.get(flow, multipleCondition.getId())
             .filter(m -> m.isValid(ZonedDateTime.now()))
diff --git a/core/src/main/java/io/kestra/core/runners/RunVariables.java b/core/src/main/java/io/kestra/core/runners/RunVariables.java
index 0ac5673c1..c348146ae 100644
--- a/core/src/main/java/io/kestra/core/runners/RunVariables.java
+++ b/core/src/main/java/io/kestra/core/runners/RunVariables.java
@@ -1,6 +1,7 @@
 package io.kestra.core.runners;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
@@ -312,10 +313,7 @@ public final class RunVariables {
                     builder.put("labels", execution.getLabels()
                         .stream()
                         .filter(label -> label.value() != null && label.key() != null)
-                        .map(label -> new AbstractMap.SimpleEntry<>(
-                            label.key(),
-                            label.value()
-                        ))
+                        .map(label -> mapLabel(label))
                         // using an accumulator in case labels with the same key exists: the first is kept
                         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (first, second) -> first))
                     );
@@ -349,5 +347,22 @@ public final class RunVariables {
         }
     }
 
+    private static Map.Entry<String, Object> mapLabel(Label label) {
+        if (label.key().startsWith(Label.SYSTEM_PREFIX)) {
+            return Map.entry(
+                label.key().substring(0, Label.SYSTEM_PREFIX.length() - 1),
+                Map.entry(
+                    label.key().substring(Label.SYSTEM_PREFIX.length()),
+                    label.value()
+                )
+            );
+        } else {
+            return Map.entry(
+                label.key(),
+                label.value()
+            );
+        }
+    }
+
     private RunVariables(){}
 }
diff --git a/core/src/main/java/io/kestra/core/validations/ExecutionFiltersConditionValidation.java b/core/src/main/java/io/kestra/core/validations/ExecutionFiltersConditionValidation.java
new file mode 100644
index 000000000..94dc0839b
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/validations/ExecutionFiltersConditionValidation.java
@@ -0,0 +1,16 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.validations.validator.ExecutionFiltersConditionValidator;
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = ExecutionFiltersConditionValidator.class)
+public @interface ExecutionFiltersConditionValidation {
+    String message() default "invalid execution filters condition";
+    Class<?>[] groups() default {};
+    Class<? extends Payload>[] payload() default {};
+}
diff --git a/core/src/main/java/io/kestra/core/validations/validator/ExecutionFiltersConditionValidator.java b/core/src/main/java/io/kestra/core/validations/validator/ExecutionFiltersConditionValidator.java
new file mode 100644
index 000000000..f025410ee
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/validations/validator/ExecutionFiltersConditionValidator.java
@@ -0,0 +1,50 @@
+package io.kestra.core.validations.validator;
+
+import io.kestra.core.validations.ExecutionFiltersConditionValidation;
+import io.kestra.plugin.core.condition.ExecutionFilters;
+import io.micronaut.core.annotation.AnnotationValue;
+import io.micronaut.core.annotation.NonNull;
+import io.micronaut.core.annotation.Nullable;
+import io.micronaut.validation.validator.constraints.ConstraintValidator;
+import io.micronaut.validation.validator.constraints.ConstraintValidatorContext;
+import jakarta.inject.Singleton;
+
+import java.util.List;
+
+@Singleton
+public class ExecutionFiltersConditionValidator  implements ConstraintValidator<ExecutionFiltersConditionValidation, ExecutionFilters.Condition> {
+    @Override
+    public boolean isValid(@Nullable ExecutionFilters.Condition value, @NonNull AnnotationValue<ExecutionFiltersConditionValidation> annotationMetadata, @NonNull ConstraintValidatorContext context) {
+        if (value == null) {
+            return true; // nulls are allowed according to spec
+        }
+
+        List<ExecutionFilters.Type> needsValue = List.of(ExecutionFilters.Type.EQUAL_TO, ExecutionFilters.Type.NOT_EQUAL_TO, ExecutionFilters.Type.IS_NULL, ExecutionFilters.Type.IS_NOT_NULL, ExecutionFilters.Type.IS_TRUE, ExecutionFilters.Type.IS_FALSE, ExecutionFilters.Type.STARTS_WITH, ExecutionFilters.Type.ENDS_WITH, ExecutionFilters.Type.REGEX, ExecutionFilters.Type.CONTAINS);
+        if (needsValue.contains(value.getType()) && value.getValue() == null) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate("`value` cannot be null for type " + value.getType())
+                .addConstraintViolation();
+            return false;
+        } else if (!needsValue.contains(value.getType()) && value.getValue() != null) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate("`value` must be null for type " + value.getType())
+                .addConstraintViolation();
+            return false;
+        }
+
+        List<ExecutionFilters.Type> needsValues = List.of(ExecutionFilters.Type.IN, ExecutionFilters.Type.NOT_IN);
+        if (needsValues.contains(value.getType()) && value.getValues() == null) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate("`values` cannot be null for type " + value.getType())
+                .addConstraintViolation();
+            return false;
+        } else if (!needsValues.contains(value.getType()) && value.getValues() != null) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate("`values` must be null for type " + value.getType())
+                .addConstraintViolation();
+            return false;
+        }
+
+        return true;
+    }
+}
diff --git a/core/src/main/java/io/kestra/plugin/core/condition/AbstractMultipleCondition.java b/core/src/main/java/io/kestra/plugin/core/condition/AbstractMultipleCondition.java
index 3c14f3808..9a9d603dc 100644
--- a/core/src/main/java/io/kestra/plugin/core/condition/AbstractMultipleCondition.java
+++ b/core/src/main/java/io/kestra/plugin/core/condition/AbstractMultipleCondition.java
@@ -32,26 +32,32 @@ public abstract class AbstractMultipleCondition extends Condition implements Mul
     @Pattern(regexp="^[a-zA-Z0-9][a-zA-Z0-9_-]*")
     @Schema(title = "A unique id for the whole flow")
     @PluginProperty
-    protected String id;
+    private String id;
+
+    @Schema(
+        title = "Latency SLA to use instead of a sliding window",
+        description = "This allow to specify at which time in a day, at least, the flow should be triggered when all conditions are met."
+    )
+    @PluginProperty
+    private LatencySLA latencySLA;
 
     @NotNull
     @Schema(
         title = "The duration of the window",
-        description = "See [ISO_8601 Durations](https://en.wikipedia.org/wiki/ISO_8601#Durations) for more information of available duration value.\n" +
-            "The start of the window is always based on midnight except if you set windowAdvance parameter. Eg if you have a 10 minutes (PT10M) window, " +
-            "the first window will be 00:00 to 00:10 and a new window will be started each 10 minutes")
+        description = """
+            See [ISO_8601 Durations](https://en.wikipedia.org/wiki/ISO_8601#Durations) for more information of available duration value.
+            The start of the window is always based on midnight except if you set windowAdvance parameter. Eg if you have a 10 minutes (PT10M) window,
+            the first window will be 00:00 to 00:10 and a new window will be started each 10 minutes.""")
     @PluginProperty
     @Builder.Default
     private Duration window = Duration.ofDays(1);
 
     @Schema(
         title = "The window advance duration",
-        description = "Allow to specify the start hour of the window\n" +
-            "Eg: you want a window of 6 hours (window=PT6H). By default the check will be done between: \n" +
-            "00:00 and 06:00 - 06:00 and 12:00 - 12:00 and 18:00 - 18:00 and 00:00 " +
-            "If you want to check the window between: \n" +
-            "03:00 and 09:00 - 09:00 and 15:00 - 15:00 and 21:00 - 21:00 and 3:00" +
-            "You will have to shift the window of 3 hours by settings windowAdvance: PT3H")
+        description = """
+            Allow to specify the start hour of the window.
+            Eg: you want a window of 6 hours (window=PT6H). By default the check will be done between: 00:00 and 06:00 - 06:00 and 12:00 - 12:00 and 18:00 - 18:00 and 00:00.
+            If you want to check the window between: 03:00 and 09:00 - 09:00 and 15:00 - 15:00 and 21:00 - 21:00 and 3:00, you will have to shift the window of 3 hours by settings windowAdvance: PT3H.""")
     @PluginProperty
     private Duration windowAdvance;
 
diff --git a/core/src/main/java/io/kestra/plugin/core/condition/ExecutionFilters.java b/core/src/main/java/io/kestra/plugin/core/condition/ExecutionFilters.java
new file mode 100644
index 000000000..ca69fc60e
--- /dev/null
+++ b/core/src/main/java/io/kestra/plugin/core/condition/ExecutionFilters.java
@@ -0,0 +1,266 @@
+package io.kestra.plugin.core.condition;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.conditions.ConditionContext;
+import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.TruthUtils;
+import io.kestra.core.validations.ExecutionFiltersConditionValidation;
+import io.swagger.v3.oas.annotations.Hidden;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+import lombok.extern.jackson.Jacksonized;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static io.kestra.core.topologies.FlowTopologyService.SIMULATED_EXECUTION;
+import static io.kestra.core.utils.Rethrow.throwPredicate;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode(callSuper = true)
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "Condition for a list of executions.",
+    description = "Trigger when all the executions are successfully executed corresponding on a set of conditions for the first time during the `window` duration."
+)
+@Plugin(
+    examples = {
+        @Example(
+            full = true,
+            title = "A flow that is waiting for 2 flows to run successfully in a day with complex filters",
+            code = """
+                id: execution-filters
+                namespace: company.team
+
+                triggers:
+                  - id: flowFilters
+                    type: io.kestra.plugin.core.trigger.Flow
+                    conditions:
+                      - id: executionFilters
+                        type: io.kestra.plugin.core.condition.ExecutionFilters
+                        filters:
+                          - id: flow1
+                            conditions:
+                              - field: NAMESPACE
+                                type: EQUAL_TO
+                                value: company.team
+                              - field: FLOW_ID
+                                type: EQUAL_TO
+                                value: myflow
+                              - field: STATE
+                                type: IN
+                                values: [SUCCESS, WARNING, CANCELLED]
+                          - id: flow2
+                            conditions:
+                              - field: NAMESPACE
+                                type: EQUAL_TO
+                                value: company.team
+                              - field: FLOW_ID
+                                type: EQUAL_TO
+                                value: myflow2
+                              - field: STATE
+                                type: EQUAL_TO
+                                values: SUCCESS
+                              - field: EXPRESSION
+                                type: IS_TRUE
+                                value: "{{outputs.output.values.variable == 'value'}}"
+                              - field: LABEL
+                                type: EQUAL_TO
+                                value: "some:label"
+
+                tasks:
+                  - id: hello
+                    type: io.kestra.plugin.core.log.Log
+                    message: I'm triggered by two flows!"""
+        )
+    }
+)
+public class ExecutionFilters extends AbstractMultipleCondition {
+    @NotNull
+    @NotEmpty
+    @Valid
+    @PluginProperty
+    @Schema(title = "The list of execution filters.")
+    private List<ExecutionFilters.Filter> filters;
+
+    /**
+     * Will emulate multiple conditions based on the filters property.
+     */
+    @JsonIgnore
+    @Override
+    public Map<String, io.kestra.core.models.conditions.Condition> getConditions() {
+        return filters.stream()
+            .map(filter -> Map.entry(
+                filter.getId(),
+                new ExecutionFilters.FilterCondition(filter)
+            ))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
+    @Hidden
+    public static class FilterCondition extends io.kestra.core.models.conditions.Condition {
+        private final ExecutionFilters.Filter filter;
+
+        private FilterCondition(ExecutionFilters.Filter filter) {
+            this.filter = Objects.requireNonNull(filter);
+        }
+
+        @Override
+        public boolean test(ConditionContext conditionContext) throws InternalException {
+            // we need to only evaluate on namespace and flow for simulated executions
+            boolean simulated = ListUtils.emptyOnNull(conditionContext.getExecution().getLabels()).contains(SIMULATED_EXECUTION);
+            Stream<Condition> toEvaluate = simulated ? filter.conditions.stream().filter(condition -> condition.field == Field.NAMESPACE || condition.field == Field.FLOW_ID) : filter.conditions.stream();
+
+            return switch (filter.operand) {
+                case AND -> toEvaluate.allMatch(throwPredicate(condition -> evaluate(conditionContext, condition)));
+                case OR -> toEvaluate.anyMatch(throwPredicate(condition -> evaluate(conditionContext, condition)));
+            };
+        }
+
+        private boolean evaluate(ConditionContext conditionContext, Condition condition) throws IllegalVariableEvaluationException {
+            String fieldValue = switch (condition.field) {
+                case FLOW_ID -> conditionContext.getExecution().getFlowId();
+                case NAMESPACE -> conditionContext.getExecution().getNamespace();
+                case STATE -> conditionContext.getExecution().getState().getCurrent().toString();
+                case EXPRESSION -> conditionContext.getRunContext().render(condition.value);
+                case LABEL -> conditionContext.getExecution().getLabels().stream()
+                    .filter(label -> label.key().equals(extractLabelKey(condition.value)))
+                    .findFirst()
+                    .map(label -> label.key() + ":" + label.value())
+                    .orElse(null);
+            };
+
+            return switch (condition.type) {
+                case EQUAL_TO -> condition.value.equals(fieldValue);
+                case NOT_EQUAL_TO -> !condition.value.equals(fieldValue);
+                case IN -> condition.values.contains(fieldValue);
+                case NOT_IN -> !condition.values.contains(fieldValue);
+                case IS_TRUE -> TruthUtils.isTruthy(fieldValue);
+                case IS_FALSE -> !TruthUtils.isTruthy(fieldValue);
+                case IS_NULL -> fieldValue == null;
+                case IS_NOT_NULL -> fieldValue != null;
+                case STARTS_WITH -> fieldValue != null && fieldValue.startsWith(condition.value);
+                case ENDS_WITH -> fieldValue != null && fieldValue.endsWith(condition.value);
+                case REGEX -> fieldValue != null && fieldValue.matches(condition.value);
+                case CONTAINS -> fieldValue != null && fieldValue.contains(condition.value);
+            };
+        }
+
+        private String extractLabelKey(String value) {
+            if (value.indexOf(':') < 0) {
+                // will allow checking the existence of a label
+                return value;
+            }
+            // will allow checking the value of a label
+            return value.substring(0, value.indexOf(':'));
+        }
+    }
+
+    @Builder
+    @Getter
+    @Jacksonized
+    public static class Filter {
+        @NotNull
+        @NotEmpty
+        @PluginProperty
+        @Schema(title = "A unique identifier for the filter")
+        private String id;
+
+        @NotNull
+        @Builder.Default
+        @PluginProperty
+        @Schema(title = "The operand to apply between all conditions of the filter.")
+        private Operand operand = Operand.AND;
+
+        @NotNull
+        @NotEmpty
+        @Valid
+        @PluginProperty
+        @Schema(title = "The list of conditions.")
+        private List<Condition> conditions;
+    }
+
+    public enum Operand {
+        AND,
+        OR
+    }
+
+    @Builder
+    @Getter
+    @ExecutionFiltersConditionValidation
+    public static class Condition {
+        @NotNull
+        @PluginProperty
+        @Schema(
+            title = "The field the condition applied to.",
+            description = """
+                Labels have a special handling, they must be matched by their value using a special syntax: `key:value`.
+                For example, to match executions that have a label with the key `productId` you can use this filter:
+                ```
+              - field: LABEL
+                type: IS_NOT_NULL
+                value: productId
+                ```"""
+        )
+        private Field field;
+
+        @NotNull
+        @PluginProperty
+        @Schema(
+            title = "The type of condition.",
+            description = "Depending on the type, you will need to also set the `value` or `values` property."
+        )
+        private Type type;
+
+        @PluginProperty
+        @Schema(
+            title = "The single value to filter the field on.",
+            description = "Must be set for the following types: EQUAL_TO, NOT_EQUAL_TO, IS_NULL, IS_NOT_NULL, IS_TRUE, IS_FALSE, STARTS_WITH, ENDS_WITH, REGEX, CONTAINS."
+        )
+        private String value;
+
+        @PluginProperty
+        @Schema(
+            title = "The list of values to filter the field on.",
+            description = "Must be set for the following types: IN, NOT_IN."
+        )
+        private List<String> values;
+    }
+
+    public enum Field {
+        FLOW_ID,
+        NAMESPACE,
+        STATE,
+        EXPRESSION,
+        LABEL,
+    }
+
+    public enum Type {
+        EQUAL_TO,
+        NOT_EQUAL_TO,
+        IN,
+        NOT_IN,
+        IS_TRUE,
+        IS_FALSE,
+        IS_NULL,
+        IS_NOT_NULL,
+        STARTS_WITH,
+        ENDS_WITH,
+        REGEX,
+        CONTAINS,
+    }
+}
diff --git a/core/src/main/java/io/kestra/plugin/core/condition/FlowFilters.java b/core/src/main/java/io/kestra/plugin/core/condition/FlowFilters.java
index c1eac7b63..b2efe9f89 100644
--- a/core/src/main/java/io/kestra/plugin/core/condition/FlowFilters.java
+++ b/core/src/main/java/io/kestra/plugin/core/condition/FlowFilters.java
@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.flows.State;
@@ -15,7 +16,6 @@ import jakarta.validation.constraints.NotEmpty;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
-import lombok.extern.slf4j.Slf4j;
 
 import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -62,10 +62,11 @@ import static io.kestra.core.topologies.FlowTopologyService.SIMULATED_EXECUTION;
         )
     }
 )
-@Slf4j
 public class FlowFilters extends AbstractMultipleCondition {
     @NotNull
     @NotEmpty
+    @Schema(title = "The list of upstream flows to wait for.")
+    @PluginProperty
     private List<UpstreamFlow> upstreamFlows;
 
     /**
diff --git a/core/src/main/java/io/kestra/plugin/core/execution/Labels.java b/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
index c633ff47b..3700fde86 100644
--- a/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
+++ b/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
@@ -37,7 +37,7 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @NoArgsConstructor
 @Schema(
     title = "Allow to add or overwrite labels for the current execution at runtime.",
-    description = "Trying to pass a system label (a label starting with `system_`) will fail the task."
+    description = "Trying to pass a system label (a label starting with `system.`) will fail the task."
 )
 @Plugin(
     examples = {
diff --git a/ui/src/components/flows/FlowRoot.vue b/ui/src/components/flows/FlowRoot.vue
index 520bd00af..f3e6e0b81 100644
--- a/ui/src/components/flows/FlowRoot.vue
+++ b/ui/src/components/flows/FlowRoot.vue
@@ -332,7 +332,7 @@
                     return false;
                 }
 
-                return this.flow.labels?.system_readOnly === "true" ?? false;
+                return this.flow.labels?.["system.readOnly"] === "true" ?? false;
             },
             routeFlowDependencies() {
                 return this.dependenciesCount > 0 ? FlowDependencies : FlowNoDependencies;
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 3aa7ca76e..542adea7a 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -688,7 +688,7 @@ public class ExecutionController {
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .toList();
 
-        // check for system labels: none can be passed at execution creation time except system_correlationId
+        // check for system labels: none can be passed at execution creation time except system.correlationId
         Optional<Label> first = parsedLabels.stream().filter(label -> !label.key().equals(CORRELATION_ID) && label.key().startsWith(SYSTEM_PREFIX)).findFirst();
         if (first.isPresent()) {
             throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().key() + "=" + first.get().value());


Test patch:
diff --git a/core/src/test/java/io/kestra/core/models/triggers/multipleflows/AbstractMultipleConditionStorageTest.java b/core/src/test/java/io/kestra/core/models/triggers/multipleflows/AbstractMultipleConditionStorageTest.java
index af36b64c9..224768c8d 100644
--- a/core/src/test/java/io/kestra/core/models/triggers/multipleflows/AbstractMultipleConditionStorageTest.java
+++ b/core/src/test/java/io/kestra/core/models/triggers/multipleflows/AbstractMultipleConditionStorageTest.java
@@ -28,6 +28,23 @@ public abstract class AbstractMultipleConditionStorageTest {
 
     abstract protected void save(MultipleConditionStorageInterface multipleConditionStorage, Flow flow, List<MultipleConditionWindow> multipleConditionWindows);
 
+    @Test
+    void allDefault() {
+        MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+
+        Pair<Flow, MultipleCondition> pair = mockFlow(null, null);
+
+        MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight());
+
+        assertThat(window.getFlowId(), is(pair.getLeft().getId()));
+
+        assertThat(window.getStart().toLocalTime(), is(LocalTime.parse("00:00:00")));
+        assertThat(window.getStart().toLocalDate(), is(ZonedDateTime.now().toLocalDate()));
+
+        assertThat(window.getEnd().toLocalTime(), is(LocalTime.parse("23:59:59.999")));
+        assertThat(window.getEnd().toLocalDate(), is(ZonedDateTime.now().toLocalDate()));
+    }
+
     @Test
     void daily() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
@@ -139,11 +156,39 @@ public abstract class AbstractMultipleConditionStorageTest {
         assertThat(expired.size(), is(1));
     }
 
+    @Test
+    void deadline() throws Exception {
+        MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+
+        Pair<Flow, MultipleCondition> pair = mockFlow(LocalTime.now().plusSeconds(1L));
+
+        MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight());
+        this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
+        assertThat(window.getFlowId(), is(pair.getLeft().getId()));
+        window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight());
+
+        assertThat(window.getResults().get("a"), is(true));
+
+        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(null);
+        assertThat(expired.size(), is(0));
+
+        Thread.sleep(1005);
+
+        expired = multipleConditionStorage.expired(null);
+        assertThat(expired.size(), is(1));
+    }
+
     private static Pair<Flow, MultipleCondition> mockFlow(Duration window, Duration advance) {
-        MultipleCondition multipleCondition = MultipleCondition.builder()
+        return mockFlow(window, advance, null);
+    }
+
+    private static Pair<Flow, MultipleCondition> mockFlow(LocalTime deadline) {
+        return mockFlow(null, null, deadline);
+    }
+
+    private static Pair<Flow, MultipleCondition> mockFlow(Duration window, Duration advance, LocalTime deadline) {
+        var multipleCondition = MultipleCondition.builder()
             .id("condition-multiple")
-            .window(window)
-            .windowAdvance(advance)
             .conditions(ImmutableMap.of(
                 "flow-a", ExecutionFlowCondition.builder()
                     .flowId("flow-a")
@@ -153,8 +198,16 @@ public abstract class AbstractMultipleConditionStorageTest {
                     .flowId("flow-b")
                     .namespace(NAMESPACE)
                     .build()
-            ))
-            .build();
+            ));
+        if (window != null) {
+            multipleCondition.window(window);
+        }
+        if (advance != null) {
+            multipleCondition.windowAdvance(advance);
+        }
+        if (deadline != null) {
+            multipleCondition.latencySLA(io.kestra.core.models.triggers.multipleflows.MultipleCondition.LatencySLA.builder().deadline(deadline).build());
+        }
 
         Flow flow = Flow.builder()
             .namespace(NAMESPACE)
@@ -162,10 +215,10 @@ public abstract class AbstractMultipleConditionStorageTest {
             .revision(1)
             .triggers(Collections.singletonList(io.kestra.plugin.core.trigger.Flow.builder()
                 .id("trigger-flow")
-                .conditions(Collections.singletonList(multipleCondition))
+                .conditions(Collections.singletonList(multipleCondition.build()))
                 .build()))
             .build();
 
-        return Pair.of(flow, multipleCondition);
+        return Pair.of(flow, multipleCondition.build());
     }
 }
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
index eceba1c80..86a8e1292 100644
--- a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
+++ b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
@@ -283,7 +283,7 @@ public abstract class AbstractFlowRepositoryTest {
     @Test
     void findByNamespace() {
         List<Flow> save = flowRepository.findByNamespace(null, "io.kestra.tests");
-        assertThat((long) save.size(), is(Helpers.FLOWS_COUNT - 20));
+        assertThat((long) save.size(), is(Helpers.FLOWS_COUNT - 23));
 
         save = flowRepository.findByNamespace(null, "io.kestra.tests2");
         assertThat((long) save.size(), is(1L));
@@ -479,7 +479,7 @@ public abstract class AbstractFlowRepositoryTest {
     @Test
     void findDistinctNamespace() {
         List<String> distinctNamespace = flowRepository.findDistinctNamespace(null);
-        assertThat((long) distinctNamespace.size(), is(7L));
+        assertThat((long) distinctNamespace.size(), is(8L));
     }
 
     @SuppressWarnings("deprecation")
diff --git a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
index b73f5d7f4..3ba010f6f 100644
--- a/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/MultipleConditionTriggerCaseTest.java
@@ -162,4 +162,37 @@ public class MultipleConditionTriggerCaseTest {
         assertThat(triggerExecution.getTaskRunList().size(), is(1));
         assertThat(triggerExecution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
+
+    public void executionFilters() throws InterruptedException, TimeoutException, QueueException {
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        AtomicReference<Execution> flowFilters = new AtomicReference<>();
+
+        Flux<Execution> receive = TestsUtils.receive(executionQueue, either -> {
+            Execution execution = either.getLeft();
+            if (execution.getState().getCurrent() == State.Type.SUCCESS && execution.getFlowId().equals("flow-trigger-execution-filters-flow-listen")) {
+                flowFilters.set(execution);
+                countDownLatch.countDown();
+            }
+        });
+
+        // first one
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests.trigger.execution-filters", "flow-trigger-execution-filters-flow-a", Duration.ofSeconds(60));
+        assertThat(execution.getTaskRunList().size(), is(1));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+
+        // second one
+        execution = runnerUtils.runOne(null, "io.kestra.tests.trigger.execution-filters", "flow-trigger-execution-filters-flow-b", Duration.ofSeconds(60));
+        assertThat(execution.getTaskRunList().size(), is(2));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+
+        // trigger is done
+        assertTrue(countDownLatch.await(10, TimeUnit.SECONDS));
+        receive.blockLast();
+        assertThat(flowFilters.get(), notNullValue());
+
+        Execution triggerExecution = flowFilters.get();
+
+        assertThat(triggerExecution.getTaskRunList().size(), is(1));
+        assertThat(triggerExecution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
 }
diff --git a/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java b/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
index 94ee3db6a..bfc678b1e 100644
--- a/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
+++ b/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
@@ -167,7 +167,7 @@ class YamlFlowParserTest {
         Flow flow = this.parse("flows/valids/minimal.yaml");
 
         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}}\"}]}"));
+        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}}\"}]}"));
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/services/LabelServiceTest.java b/core/src/test/java/io/kestra/core/services/LabelServiceTest.java
index cb28162e8..42f0a78ff 100644
--- a/core/src/test/java/io/kestra/core/services/LabelServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/LabelServiceTest.java
@@ -25,7 +25,7 @@ class LabelServiceTest {
     @Test
     void shouldFilterSystemLabels() {
         Flow flow = Flow.builder()
-            .labels(List.of(new Label("key", "value"), new Label("system_Label", "systemValue")))
+            .labels(List.of(new Label("key", "value"), new Label(Label.SYSTEM_PREFIX + "label", "systemValue")))
             .build();
 
         List<Label> labels = LabelService.labelsExcludingSystem(flow);
@@ -38,7 +38,7 @@ class LabelServiceTest {
     void shouldReturnLabelsFromFlowAndTrigger() {
         RunContext runContext = runContextFactory.of(Map.of("variable", "variableValue"));
         Flow flow = Flow.builder()
-            .labels(List.of(new Label("key", "value"), new Label("system_Label", "systemValue")))
+            .labels(List.of(new Label("key", "value"), new Label(Label.SYSTEM_PREFIX + "label", "systemValue")))
             .build();
         AbstractTrigger trigger = Schedule.builder()
             .labels(List.of(new Label("scheduleLabel", "scheduleValue"), new Label("variable", "{{variable}}")))
@@ -54,7 +54,7 @@ class LabelServiceTest {
     void shouldFilterNonRenderableLabels() {
         RunContext runContext = runContextFactory.of();
         Flow flow = Flow.builder()
-            .labels(List.of(new Label("key", "value"), new Label("system_Label", "systemValue")))
+            .labels(List.of(new Label("key", "value"), new Label(Label.SYSTEM_PREFIX + "label", "systemValue")))
             .build();
         AbstractTrigger trigger = Schedule.builder()
             .labels(List.of(new Label("scheduleLabel", "scheduleValue"), new Label("variable", "{{variable}}")))
diff --git a/core/src/test/java/io/kestra/core/validations/ExecutionFiltersConditionValidationTest.java b/core/src/test/java/io/kestra/core/validations/ExecutionFiltersConditionValidationTest.java
new file mode 100644
index 000000000..82d6dd8cc
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/validations/ExecutionFiltersConditionValidationTest.java
@@ -0,0 +1,94 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.models.validations.ModelValidator;
+import io.kestra.plugin.core.condition.ExecutionFilters;
+import jakarta.inject.Inject;
+import jakarta.validation.ConstraintViolationException;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+@KestraTest
+class ExecutionFiltersConditionValidationTest {
+    @Inject
+    private ModelValidator modelValidator;
+
+    @ParameterizedTest
+    @EnumSource(value = ExecutionFilters.Type.class, names = {"EQUAL_TO", "NOT_EQUAL_TO", "IS_NULL", "IS_NOT_NULL", "IS_TRUE", "IS_FALSE", "STARTS_WITH", "ENDS_WITH", "REGEX", "CONTAINS"})
+    void shouldValidateConditionWithAValue(ExecutionFilters.Type type) {
+        var condition = ExecutionFilters.Condition.builder()
+            .field(ExecutionFilters.Field.FLOW_ID)
+            .type(type)
+            .value("myFlow")
+            .build();
+
+        Optional<ConstraintViolationException> valid = modelValidator.isValid(condition);
+        assertThat(valid.isEmpty(), is(true));
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = ExecutionFilters.Type.class, names = {"EQUAL_TO", "NOT_EQUAL_TO", "IS_NULL", "IS_NOT_NULL", "IS_TRUE", "IS_FALSE", "STARTS_WITH", "ENDS_WITH", "REGEX", "CONTAINS"})
+    void shouldNotValidateConditionWithValues(ExecutionFilters.Type type) {
+        var condition = ExecutionFilters.Condition.builder()
+            .field(ExecutionFilters.Field.FLOW_ID)
+            .type(type)
+            .values(List.of("myFlow1", "myFlow2"))
+            .build();
+
+        Optional<ConstraintViolationException> valid = modelValidator.isValid(condition);
+        assertThat(valid.isEmpty(), is(false));
+        assertThat(valid.get().getConstraintViolations(), hasSize(1));
+        assertThat(valid.get().getMessage(), is(": `value` cannot be null for type " + type.name() + "\n"));
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = ExecutionFilters.Type.class, names = {"EQUAL_TO", "NOT_EQUAL_TO", "IS_NULL", "IS_NOT_NULL", "IS_TRUE", "IS_FALSE", "STARTS_WITH", "ENDS_WITH", "REGEX", "CONTAINS"})
+    void shouldNotValidateConditionWithAValueAndValues(ExecutionFilters.Type type) {
+        var condition = ExecutionFilters.Condition.builder()
+            .field(ExecutionFilters.Field.FLOW_ID)
+            .type(type)
+            .value("myFlow")
+            .values(List.of("myFlow1", "myFlow2"))
+            .build();
+
+        Optional<ConstraintViolationException> valid = modelValidator.isValid(condition);
+        assertThat(valid.isEmpty(), is(false));
+        assertThat(valid.get().getConstraintViolations(), hasSize(1));
+        assertThat(valid.get().getMessage(), is(": `values` must be null for type " + type.name() + "\n"));
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = ExecutionFilters.Type.class, names = {"IN", "NOT_IN"})
+    void shouldValidateConditionWithValues(ExecutionFilters.Type type) {
+        var condition = ExecutionFilters.Condition.builder()
+            .field(ExecutionFilters.Field.FLOW_ID)
+            .type(type)
+            .values(List.of("myFlow1", "myFlow2"))
+            .build();
+
+        Optional<ConstraintViolationException> valid = modelValidator.isValid(condition);
+        assertThat(valid.isEmpty(), is(true));
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = ExecutionFilters.Type.class, names = {"IN", "NOT_IN"})
+    void shouldNotValidateConditionWithAValue(ExecutionFilters.Type type) {
+        var condition = ExecutionFilters.Condition.builder()
+            .field(ExecutionFilters.Field.FLOW_ID)
+            .type(type)
+            .value("myFlow1")
+            .build();
+
+        Optional<ConstraintViolationException> valid = modelValidator.isValid(condition);
+        assertThat(valid.isEmpty(), is(false));
+        assertThat(valid.get().getConstraintViolations(), hasSize(1));
+        assertThat(valid.get().getMessage(), is(": `value` must be null for type " + type.name() + "\n"));
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java b/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
index b92f343f4..3866e67ba 100644
--- a/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
+++ b/core/src/test/java/io/kestra/core/validations/FlowValidationTest.java
@@ -39,8 +39,8 @@ class FlowValidationTest {
         Optional<ConstraintViolationException> validate = modelValidator.isValid(flow);
 
         assertThat(validate.isPresent(), is(true));
-        assertThat(validate.get().getMessage(), containsString("System labels can only be set by Kestra itself, offending label: system_label=system_key"));
-        assertThat(validate.get().getMessage(), containsString("System labels can only be set by Kestra itself, offending label: system_id=id"));
+        assertThat(validate.get().getMessage(), containsString("System labels can only be set by Kestra itself, offending label: system.label=system_key"));
+        assertThat(validate.get().getMessage(), containsString("System labels can only be set by Kestra itself, offending label: system.id=id"));
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/RuntimeLabelsTest.java b/core/src/test/java/io/kestra/plugin/core/flow/RuntimeLabelsTest.java
index 05fc8ccf0..0c7afe3f4 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/RuntimeLabelsTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/RuntimeLabelsTest.java
@@ -41,7 +41,7 @@ class RuntimeLabelsTest extends AbstractMemoryRunnerTest {
 
         String labelsOverriderTaskRunId = execution.findTaskRunsByTaskId("override-labels").getFirst().getId();
         assertThat(execution.getLabels(), containsInAnyOrder(
-            is(new Label("system_correlationId", execution.getId())),
+            is(new Label(Label.CORRELATION_ID, execution.getId())),
             is(new Label("flowLabelKey", "flowLabelValue")),
             is(new Label("overriddenFlowLabelKey", "io.kestra.tests.labels-update-task")),
             is(new Label("keyFromJson", "valueFromJson")),
diff --git a/core/src/test/resources/flows/invalids/system-labels.yaml b/core/src/test/resources/flows/invalids/system-labels.yaml
index 8f8ff6292..acc6b2a7b 100644
--- a/core/src/test/resources/flows/invalids/system-labels.yaml
+++ b/core/src/test/resources/flows/invalids/system-labels.yaml
@@ -2,8 +2,8 @@ id: system-labels
 namespace: io.kestra.tests
 
 labels:
-  system_label: system_key
-  system_id: id
+  system.label: system_key
+  system.id: id
   another: label
 
 tasks:
diff --git a/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-a.yaml b/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-a.yaml
new file mode 100644
index 000000000..08510d1b9
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-a.yaml
@@ -0,0 +1,7 @@
+id: flow-trigger-execution-filters-flow-a
+namespace: io.kestra.tests.trigger.execution-filters
+
+tasks:
+  - id: only
+    type: io.kestra.plugin.core.debug.Return
+    format: "from parents: {{execution.id}}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-b.yaml b/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-b.yaml
new file mode 100644
index 000000000..d99aee352
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-b.yaml
@@ -0,0 +1,15 @@
+id: flow-trigger-execution-filters-flow-b
+namespace: io.kestra.tests.trigger.execution-filters
+
+labels:
+  some: label
+  custom_correlationId: custom
+
+tasks:
+  - id: only
+    type: io.kestra.plugin.core.debug.Return
+    format: "from parents: {{execution.id}}"
+  - id: output
+    type: io.kestra.plugin.core.output.OutputValues
+    values:
+      variable: toto
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-listen.yaml b/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-listen.yaml
new file mode 100644
index 000000000..d66434258
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-trigger-execution-filters-flow-listen.yaml
@@ -0,0 +1,66 @@
+id: flow-trigger-execution-filters-flow-listen
+namespace: io.kestra.tests.trigger.execution-filters
+
+triggers:
+  - id: flow
+    type: io.kestra.plugin.core.trigger.Flow
+    conditions:
+      - id: executionFilters
+        type: io.kestra.plugin.core.condition.ExecutionFilters
+        filters:
+          - id: flow1
+            conditions:
+              - field: NAMESPACE
+                type: EQUAL_TO
+                value: io.kestra.tests.trigger.execution-filters
+              - field: FLOW_ID
+                type: EQUAL_TO
+                value: flow-trigger-execution-filters-flow-a
+              - field: STATE
+                type: IN
+                values: [SUCCESS, WARNING, CANCELLED]
+              - field: STATE
+                type: NOT_IN
+                values: [FAILED, CANCELLED]
+          - id: flow2
+            conditions:
+              - field: NAMESPACE
+                type: EQUAL_TO
+                value: io.kestra.tests.trigger.execution-filters
+              - field: FLOW_ID
+                type: EQUAL_TO
+                value: flow-trigger-execution-filters-flow-b
+              - field: EXPRESSION
+                type: IS_TRUE
+                value: "{{outputs.output.values.variable == 'toto'}}"
+              - field: EXPRESSION
+                type: IS_FALSE
+                value: "{{outputs.output.values.variable == 'titi'}}"
+              - field: LABEL
+                type: EQUAL_TO
+                value: "some:label"
+              - field: LABEL
+                type: NOT_EQUAL_TO
+                value: "other:label"
+              - field: FLOW_ID
+                type: REGEX
+                value: .*trigger-execution-filters.*
+              - field: FLOW_ID
+                type: CONTAINS
+                value: trigger-execution-filters
+              - field: FLOW_ID
+                type: STARTS_WITH
+                value: flow-trigger-execution
+              - field: FLOW_ID
+                type: ENDS_WITH
+                value: filters-flow-b
+              - field: LABEL
+                type: IS_NULL
+                value: other
+              - field: LABEL
+                type: IS_NOT_NULL
+                value: some
+tasks:
+  - id: only
+    type: io.kestra.plugin.core.debug.Return
+    format: "It works"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/minimal.yaml b/core/src/test/resources/flows/valids/minimal.yaml
index c8e99bfbf..51e78582d 100644
--- a/core/src/test/resources/flows/valids/minimal.yaml
+++ b/core/src/test/resources/flows/valids/minimal.yaml
@@ -3,7 +3,7 @@ namespace: io.kestra.tests
 revision: 2
 
 labels:
-  system_readOnly: "true"
+  system.readOnly: "true"
 
 tasks:
 - id: date
diff --git a/core/src/test/resources/flows/valids/subflow-grand-child.yaml b/core/src/test/resources/flows/valids/subflow-grand-child.yaml
index 09e1ca5a8..1a136efed 100644
--- a/core/src/test/resources/flows/valids/subflow-grand-child.yaml
+++ b/core/src/test/resources/flows/valids/subflow-grand-child.yaml
@@ -4,4 +4,4 @@ namespace: io.kestra.tests
 tasks:
   - id: firstLevel
     type: io.kestra.plugin.core.log.Log
-    message: "My grandparent is {{labels.system_correlationId}}"
\ No newline at end of file
+    message: "My grandparent is {{labels.system.correlationId}}"
\ No newline at end of file
diff --git a/jdbc-h2/src/test/java/io/kestra/runner/h2/H2MultipleConditionStorageTest.java b/jdbc-h2/src/test/java/io/kestra/runner/h2/H2MultipleConditionStorageTest.java
index a1bf1dcb7..020bf24b7 100644
--- a/jdbc-h2/src/test/java/io/kestra/runner/h2/H2MultipleConditionStorageTest.java
+++ b/jdbc-h2/src/test/java/io/kestra/runner/h2/H2MultipleConditionStorageTest.java
@@ -6,7 +6,6 @@ import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInte
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.jdbc.JdbcTestUtils;
 import io.kestra.repository.h2.H2Repository;
-import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import org.junit.jupiter.api.BeforeEach;
@@ -14,9 +13,6 @@ import org.junit.jupiter.api.BeforeEach;
 import java.util.List;
 
 class H2MultipleConditionStorageTest extends AbstractMultipleConditionStorageTest {
-    @Inject
-    ApplicationContext applicationContext;
-
     @Inject
     JdbcTestUtils jdbcTestUtils;
 
@@ -29,7 +25,7 @@ class H2MultipleConditionStorageTest extends AbstractMultipleConditionStorageTes
     }
 
     protected void save(MultipleConditionStorageInterface multipleConditionStorage, Flow flow, List<MultipleConditionWindow> multipleConditionWindows) {
-        ((H2MultipleConditionStorage) multipleConditionStorage).save(multipleConditionWindows);
+        multipleConditionStorage.save(multipleConditionWindows);
     }
 
 
diff --git a/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlMultipleConditionStorageTest.java b/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlMultipleConditionStorageTest.java
index 4c8d13f33..ddd907084 100644
--- a/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlMultipleConditionStorageTest.java
+++ b/jdbc-mysql/src/test/java/io/kestra/runner/mysql/MysqlMultipleConditionStorageTest.java
@@ -6,7 +6,6 @@ import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInte
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.jdbc.JdbcTestUtils;
 import io.kestra.repository.mysql.MysqlRepository;
-import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import org.junit.jupiter.api.BeforeEach;
@@ -14,9 +13,6 @@ import org.junit.jupiter.api.BeforeEach;
 import java.util.List;
 
 class MysqlMultipleConditionStorageTest extends AbstractMultipleConditionStorageTest {
-    @Inject
-    ApplicationContext applicationContext;
-
     @Inject
     JdbcTestUtils jdbcTestUtils;
 
@@ -29,7 +25,7 @@ class MysqlMultipleConditionStorageTest extends AbstractMultipleConditionStorage
     }
 
     protected void save(MultipleConditionStorageInterface multipleConditionStorage, Flow flow, List<MultipleConditionWindow> multipleConditionWindows) {
-        ((MysqlMultipleConditionStorage) multipleConditionStorage).save(multipleConditionWindows);
+        multipleConditionStorage.save(multipleConditionWindows);
     }
 
 
diff --git a/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresMultipleConditionStorageTest.java b/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresMultipleConditionStorageTest.java
index e89df1b41..26b66e9cf 100644
--- a/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresMultipleConditionStorageTest.java
+++ b/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresMultipleConditionStorageTest.java
@@ -6,7 +6,6 @@ import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInte
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.jdbc.JdbcTestUtils;
 import io.kestra.repository.postgres.PostgresRepository;
-import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import org.junit.jupiter.api.BeforeEach;
@@ -14,9 +13,6 @@ import org.junit.jupiter.api.BeforeEach;
 import java.util.List;
 
 class PostgresMultipleConditionStorageTest extends AbstractMultipleConditionStorageTest {
-    @Inject
-    ApplicationContext applicationContext;
-
     @Inject
     JdbcTestUtils jdbcTestUtils;
 
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 c67a9a888..18aaab8ab 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -232,6 +232,11 @@ public abstract class JdbcRunnerTest {
         multipleConditionTriggerCaseTest.flowFilters();
     }
 
+    @Test
+    void multipleConditionTriggerExecutionFilters() throws Exception {
+        multipleConditionTriggerCaseTest.executionFilters();
+    }
+
     @RetryingTest(5)
     void eachWithNull() throws Exception {
         EachSequentialTest.eachNullTest(runnerUtils, logsQueue);
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
index 5ac32d003..0799798e2 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
@@ -1597,7 +1597,7 @@ class ExecutionControllerTest extends JdbcH2ControllerTest {
     void shouldRefuseSystemLabelsWhenCreatingAnExecution() {
         var error = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().retrieve(
             HttpRequest
-                .POST("/api/v1/executions/io.kestra.tests/minimal?labels=system_label:system", null)
+                .POST("/api/v1/executions/io.kestra.tests/minimal?labels=system.label:system", null)
                 .contentType(MediaType.MULTIPART_FORM_DATA_TYPE),
             Execution.class
         ));
@@ -1612,7 +1612,7 @@ class ExecutionControllerTest extends JdbcH2ControllerTest {
         assertThat(result.getState().getCurrent(), is(State.Type.SUCCESS));
 
         var error = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().retrieve(
-                HttpRequest.POST("/api/v1/executions/" + result.getId() + "/labels", List.of(new Label("system_label", "value"))),
+                HttpRequest.POST("/api/v1/executions/" + result.getId() + "/labels", List.of(new Label("system.label", "value"))),
                 Execution.class
             )
         );
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/FlowControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/FlowControllerTest.java
index 5efc20a97..3c6ada3dc 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/FlowControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/FlowControllerTest.java
@@ -433,7 +433,7 @@ class FlowControllerTest extends JdbcH2ControllerTest {
         List<String> namespaces = client.toBlocking().retrieve(
             HttpRequest.GET("/api/v1/flows/distinct-namespaces"), Argument.listOf(String.class));
 
-        assertThat(namespaces.size(), is(7));
+        assertThat(namespaces.size(), is(8));
     }
 
     @Test