chore(version): bump ui-libs for a fix. (#12425)

Co-authored-by: Piyush Bhaskar <impiyush0012@gmail.com>
Co-authored-by: Piyush Bhaskar <102078527+Piyush-r-bhaskar@users.noreply.github.com>

.github/CONTRIBUTING.md

@@ -32,7 +32,7 @@ Watch out for duplicates! If you are creating a new issue, please check existing
 #### Requirements
 The following dependencies are required to build Kestra locally:
 - Java 21+
-- Node 18+ and npm
+- Node 22+ and npm 10+
 - Python 3, pip and python venv
 - Docker & Docker Compose
 - an IDE (Intellij IDEA, Eclipse or VS Code)

.github/dependabot.yml

@@ -26,6 +26,10 @@ updates:
     open-pull-requests-limit: 50
     labels:
       - "dependency-upgrade"
+    ignore:
+      - dependency-name: "com.google.protobuf:*"
+        # Ignore versions of Protobuf that are equal to or greater than 4.0.0 as Orc still uses 3
+        versions: [ "[4,)" ]
 
   # Maintain dependencies for NPM modules
   - package-ecosystem: "npm"

.github/workflows/auto-translate-ui-keys.yml

@@ -2,7 +2,7 @@ name: Auto-Translate UI keys and create PR
 
 on:
   schedule:
-    - cron: "0 9-21/3 * * *"  # Every 3 hours from 9 AM to 9 PM
+    - cron: "0 9-21/3 * * 1-5"  # Every 3 hours from 9 AM to 9 PM, Monday to Friday
   workflow_dispatch:
     inputs:
       retranslate_modified_keys:
@@ -39,7 +39,7 @@ jobs:
             OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
 
         - name: Set up Node
-          uses: actions/setup-node@v5
+          uses: actions/setup-node@v6
           with:
             node-version: "20.x"
 

.github/workflows/main-build.yml

@@ -68,20 +68,23 @@ jobs:
   end:
     runs-on: ubuntu-latest
     needs: [backend-tests, frontend-tests, publish-develop-docker, publish-develop-maven]
-    if: always()
+    if: "always() && github.repository == 'kestra-io/kestra'"
     steps:
-      - run: echo "debug repo ${{github.repository}} ref ${{github.ref}} res ${{needs.publish-develop-maven.result}} jobStatus ${{job.status}} isNotFork ${{github.repository == 'kestra-io/kestra'}} isDevelop ${{github.ref == 'refs/heads/develop'}}"
+      - run: echo "end CI of failed or success"
       - name: Trigger EE Workflow
-        uses: peter-evans/repository-dispatch@v4
-        if: github.ref == 'refs/heads/develop' && needs.publish-develop-maven == 'success'
+        uses: peter-evans/repository-dispatch@5fc4efd1a4797ddb68ffd0714a238564e4cc0e6f # v4
+        if: "!contains(needs.*.result, 'failure') && github.ref == 'refs/heads/develop'"
         with:
           token: ${{ secrets.GH_PERSONAL_TOKEN }}
           repository: kestra-io/kestra-ee
           event-type: "oss-updated"
 
       # Slack
+      - run: echo "mark job as failure to forward error to Slack action" && exit 1
+        if: ${{ contains(needs.*.result, 'failure') }}
       - name: Slack - Notification
-        if: ${{ failure() && github.repository == 'kestra-io/kestra' && (github.ref == 'refs/heads/develop') }}
+        if: ${{ always() && contains(needs.*.result, 'failure') }}
         uses: kestra-io/actions/composite/slack-status@main
         with:
           webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}
+          channel: 'C09FF36GKE1'

.github/workflows/vulnerabilities-check.yml

@@ -71,7 +71,7 @@ jobs:
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.33.1
+        uses: aquasecurity/trivy-action@b6643a29fecd7f34b3597bc6acb0a98b03d33ff8 # 0.33.1
         with:
           image-ref: kestra/kestra:develop
           format: 'template'
@@ -108,7 +108,7 @@ jobs:
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.33.1
+        uses: aquasecurity/trivy-action@b6643a29fecd7f34b3597bc6acb0a98b03d33ff8 # 0.33.1
         with:
           image-ref: kestra/kestra:latest
           format: table

.plugins

@@ -66,6 +66,7 @@
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-sybase:LATEST
 #plugin-jenkins:io.kestra.plugin:plugin-jenkins:LATEST
 #plugin-jira:io.kestra.plugin:plugin-jira:LATEST
+#plugin-jms:io.kestra.plugin:plugin-jms:LATEST
 #plugin-kafka:io.kestra.plugin:plugin-kafka:LATEST
 #plugin-kestra:io.kestra.plugin:plugin-kestra:LATEST
 #plugin-kubernetes:io.kestra.plugin:plugin-kubernetes:LATEST

README.md

@@ -98,7 +98,7 @@ If you're on Windows and use WSL (Linux-based environment in Windows):
 ```bash
 docker run --pull=always --rm -it -p 8080:8080 --user=root \
     -v "/var/run/docker.sock:/var/run/docker.sock" \
-    -v "C:/Temp:/tmp" kestra/kestra:latest server local
+    -v "/mnt/c/Temp:/tmp" kestra/kestra:latest server local
 ```
 
 Check our [Installation Guide](https://kestra.io/docs/installation) for other deployment options (Docker Compose, Podman, Kubernetes, AWS, GCP, Azure, and more).

build.gradle

@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "6.3.1.5724"
+    id "org.sonarqube" version "7.0.0.6105"
     id 'jacoco-report-aggregation'
 
     // helper
@@ -37,7 +37,7 @@ plugins {
     id "com.vanniktech.maven.publish" version "0.34.0"
 
     // OWASP dependency check
-    id "org.owasp.dependencycheck" version "12.1.6" apply false
+    id "org.owasp.dependencycheck" version "12.1.8" apply false
 }
 
 idea {
@@ -372,7 +372,7 @@ tasks.named('testCodeCoverageReport') {
 subprojects {
     sonar {
         properties {
-            property "sonar.coverage.jacoco.xmlReportPaths", "$projectDir.parentFile.path/build/reports/jacoco/testCodeCoverageReport/testCodeCoverageReport.xml,$projectDir.parentFile.path/build/reports/jacoco/test//testCodeCoverageReport.xml"
+            property "sonar.coverage.jacoco.xmlReportPaths", "$projectDir.parentFile.path/build/reports/jacoco/testCodeCoverageReport/testCodeCoverageReport.xml,$projectDir.parentFile.path/build/reports/jacoco/test/testCodeCoverageReport.xml"
         }
     }
 }

cli/src/main/java/io/kestra/cli/AbstractValidateCommand.java

@@ -117,7 +117,7 @@ public abstract class AbstractValidateCommand extends AbstractApiCommand {
 
             try(DefaultHttpClient client = client()) {
                 MutableHttpRequest<String> request = HttpRequest
-                    .POST(apiUri("/flows/validate", tenantService.getTenantId(tenantId)), body).contentType(MediaType.APPLICATION_YAML);
+                    .POST(apiUri("/flows/validate", tenantService.getTenantIdAndAllowEETenants(tenantId)), body).contentType(MediaType.APPLICATION_YAML);
 
                 List<ValidateConstraintViolation> validations = client.toBlocking().retrieve(
                     this.requestOptions(request),

cli/src/main/java/io/kestra/cli/commands/flows/FlowValidateCommand.java

@@ -24,7 +24,8 @@ public class FlowValidateCommand extends AbstractValidateCommand {
     private FlowService flowService;
 
     @Inject
-    private TenantIdSelectorService tenantService;
+    private TenantIdSelectorService tenantIdSelectorService;
+
 
     @Override
     public Integer call() throws Exception {
@@ -39,7 +40,7 @@ public class FlowValidateCommand extends AbstractValidateCommand {
                 FlowWithSource flow = (FlowWithSource) object;
                 List<String> warnings = new ArrayList<>();
                 warnings.addAll(flowService.deprecationPaths(flow).stream().map(deprecation -> deprecation + " is deprecated").toList());
-                warnings.addAll(flowService.warnings(flow, tenantService.getTenantId(tenantId)));
+                warnings.addAll(flowService.warnings(flow, tenantIdSelectorService.getTenantIdAndAllowEETenants(tenantId)));
                 return warnings;
             },
             (Object object) -> {

cli/src/main/java/io/kestra/cli/commands/flows/namespaces/FlowNamespaceUpdateCommand.java

@@ -64,7 +64,7 @@ public class FlowNamespaceUpdateCommand extends AbstractServiceNamespaceUpdateCo
             }
             try(DefaultHttpClient client = client()) {
                 MutableHttpRequest<String> request = HttpRequest
-                    .POST(apiUri("/flows/", tenantService.getTenantId(tenantId)) + namespace + "?delete=" + delete, body).contentType(MediaType.APPLICATION_YAML);
+                    .POST(apiUri("/flows/", tenantService.getTenantIdAndAllowEETenants(tenantId)) + namespace + "?delete=" + delete, body).contentType(MediaType.APPLICATION_YAML);
 
                 List<UpdateResult> updated = client.toBlocking().retrieve(
                     this.requestOptions(request),

cli/src/main/java/io/kestra/cli/commands/migrations/MetadataMigrationCommand.java

@@ -0,0 +1,28 @@
+package io.kestra.cli.commands.migrations;
+
+import io.kestra.cli.AbstractCommand;
+import jakarta.inject.Inject;
+import lombok.extern.slf4j.Slf4j;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+    name = "metadata",
+    description = "populate metadata for entities"
+)
+@Slf4j
+public class MetadataMigrationCommand extends AbstractCommand {
+    @Inject
+    private MetadataMigrationService metadataMigrationService;
+
+    @Override
+    public Integer call() throws Exception {
+        super.call();
+        try {
+            metadataMigrationService.migrateMetadata();
+            System.out.println("✅ Metadata migration complete.");
+            return 0;
+        } catch (Exception e) {
+            return 1;
+        }
+    }
+}

cli/src/main/java/io/kestra/cli/commands/migrations/MetadataMigrationService.java

@@ -0,0 +1,11 @@
+package io.kestra.cli.commands.migrations;
+
+import jakarta.inject.Singleton;
+
+@Singleton
+public class MetadataMigrationService {
+    public int migrateMetadata() {
+        // no-op
+        return 0;
+    }
+}

cli/src/main/java/io/kestra/cli/commands/migrations/MigrationCommand.java

@@ -13,6 +13,7 @@ import picocli.CommandLine;
     mixinStandardHelpOptions = true,
     subcommands = {
         TenantMigrationCommand.class,
+        MetadataMigrationCommand.class
     }
 )
 @Slf4j

cli/src/main/java/io/kestra/cli/commands/namespaces/files/NamespaceFilesUpdateCommand.java

@@ -49,7 +49,7 @@ public class NamespaceFilesUpdateCommand extends AbstractApiCommand {
 
         try (var files = Files.walk(from); DefaultHttpClient client = client()) {
             if (delete) {
-                client.toBlocking().exchange(this.requestOptions(HttpRequest.DELETE(apiUri("/namespaces/", tenantService.getTenantId(tenantId)) + namespace + "/files?path=" + to, null)));
+                client.toBlocking().exchange(this.requestOptions(HttpRequest.DELETE(apiUri("/namespaces/", tenantService.getTenantIdAndAllowEETenants(tenantId)) + namespace + "/files?path=" + to, null)));
             }
 
             KestraIgnore kestraIgnore = new KestraIgnore(from);
@@ -67,7 +67,7 @@ public class NamespaceFilesUpdateCommand extends AbstractApiCommand {
                 client.toBlocking().exchange(
                     this.requestOptions(
                         HttpRequest.POST(
-                            apiUri("/namespaces/", tenantService.getTenantId(tenantId)) + namespace + "/files?path=" + destination,
+                            apiUri("/namespaces/", tenantService.getTenantIdAndAllowEETenants(tenantId)) + namespace + "/files?path=" + destination,
                             body
                         ).contentType(MediaType.MULTIPART_FORM_DATA)
                     )

cli/src/main/java/io/kestra/cli/commands/templates/namespaces/TemplateNamespaceUpdateCommand.java

@@ -49,7 +49,7 @@ public class TemplateNamespaceUpdateCommand extends AbstractServiceNamespaceUpda
 
             try (DefaultHttpClient client = client()) {
                 MutableHttpRequest<List<Template>> request = HttpRequest
-                    .POST(apiUri("/templates/", tenantService.getTenantId(tenantId)) + namespace + "?delete=" + delete, templates);
+                    .POST(apiUri("/templates/", tenantService.getTenantIdAndAllowEETenants(tenantId)) + namespace + "?delete=" + delete, templates);
 
                 List<UpdateResult> updated = client.toBlocking().retrieve(
                     this.requestOptions(request),

cli/src/main/java/io/kestra/cli/listeners/GracefulEmbeddedServiceShutdownListener.java

@@ -0,0 +1,69 @@
+package io.kestra.cli.listeners;
+
+import io.kestra.core.server.LocalServiceState;
+import io.kestra.core.server.Service;
+import io.kestra.core.server.ServiceRegistry;
+import io.micronaut.context.annotation.Requires;
+import io.micronaut.context.event.ApplicationEventListener;
+import io.micronaut.context.event.ShutdownEvent;
+import io.micronaut.core.annotation.Order;
+import io.micronaut.core.order.Ordered;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Global application shutdown handler.
+ * This handler gets effectively invoked before {@link jakarta.annotation.PreDestroy} does.
+ */
+@Singleton
+@Slf4j
+@Order(Ordered.LOWEST_PRECEDENCE)
+@Requires(property = "kestra.server-type")
+public class GracefulEmbeddedServiceShutdownListener implements ApplicationEventListener<ShutdownEvent> {
+    @Inject
+    ServiceRegistry serviceRegistry;
+
+    /**
+     * {@inheritDoc}
+     **/
+    @Override
+    public boolean supports(ShutdownEvent event) {
+        return ApplicationEventListener.super.supports(event);
+    }
+
+    /**
+     * Wait for services' close actions
+     *
+     * @param event the event to respond to
+     */
+    @Override
+    public void onApplicationEvent(ShutdownEvent event) {
+        List<LocalServiceState> states = serviceRegistry.all();
+        if (states.isEmpty()) {
+            return;
+        }
+
+        log.debug("Shutdown event received");
+
+        List<CompletableFuture<Void>> futures = states.stream()
+            .map(state -> CompletableFuture.runAsync(() -> closeService(state), ForkJoinPool.commonPool()))
+            .toList();
+
+        // Wait for all services to close, before shutting down the embedded server
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+    }
+
+    private void closeService(LocalServiceState state) {
+        final Service service = state.service();
+        try {
+            service.unwrap().close();
+        } catch (Exception e) {
+            log.error("[Service id={}, type={}] Unexpected error on close", service.getId(), service.getType(), e);
+        }
+    }
+}

cli/src/main/java/io/kestra/cli/services/TenantIdSelectorService.java

@@ -16,4 +16,11 @@ public class TenantIdSelectorService {
         }
         return MAIN_TENANT;
     }
+
+    public String getTenantIdAndAllowEETenants(String tenantId) {
+        if (StringUtils.isNotBlank(tenantId)){
+            return tenantId;
+        }
+        return MAIN_TENANT;
+    }
 }

cli/src/test/java/io/kestra/cli/commands/flows/FlowValidateCommandTest.java

@@ -27,6 +27,26 @@ class FlowValidateCommandTest {
         }
     }
 
+    @Test
+     // github action kestra-io/validate-action requires being able to validate Flows from OSS CLI against a remote EE instance
+    void runForEEInstance() {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.builder().deduceEnvironment(false).start()) {
+            String[] args = {
+                "--tenant",
+                "some-ee-tenant",
+                "--local",
+                "src/test/resources/helper/include.yaml"
+            };
+            Integer call = PicocliRunner.call(FlowValidateCommand.class, ctx, args);
+
+            assertThat(call).isZero();
+            assertThat(out.toString()).contains("✓ - io.kestra.cli / include");
+        }
+    }
+
     @Test
     void warning() {
         ByteArrayOutputStream out = new ByteArrayOutputStream();

core/src/main/java/io/kestra/core/app/AppBlockInterface.java

@@ -7,6 +7,8 @@ import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 /**
  * Top-level marker interface for Kestra's plugin of type App.
  */
@@ -18,6 +20,6 @@ public interface AppBlockInterface extends io.kestra.core.models.Plugin {
     )
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     String getType();
 }

core/src/main/java/io/kestra/core/app/AppPluginInterface.java

@@ -7,6 +7,8 @@ import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 /**
  * Top-level marker interface for Kestra's plugin of type App.
  */
@@ -18,6 +20,6 @@ public interface AppPluginInterface extends io.kestra.core.models.Plugin {
     )
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     String getType();
 }

core/src/main/java/io/kestra/core/docs/JsonSchemaGenerator.java

@@ -15,6 +15,7 @@ import com.github.victools.jsonschema.generator.impl.DefinitionKey;
 import com.github.victools.jsonschema.generator.naming.DefaultSchemaDefinitionNamingStrategy;
 import com.github.victools.jsonschema.module.jackson.JacksonModule;
 import com.github.victools.jsonschema.module.jackson.JacksonOption;
+import com.github.victools.jsonschema.module.jackson.JsonUnwrappedDefinitionProvider;
 import com.github.victools.jsonschema.module.jakarta.validation.JakartaValidationModule;
 import com.github.victools.jsonschema.module.jakarta.validation.JakartaValidationOption;
 import com.github.victools.jsonschema.module.swagger2.Swagger2Module;
@@ -45,6 +46,9 @@ import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.lang.reflect.*;
 import java.time.*;
@@ -58,7 +62,9 @@ import static io.kestra.core.docs.AbstractClassDocumentation.required;
 import static io.kestra.core.serializers.JacksonMapper.MAP_TYPE_REFERENCE;
 
 @Singleton
+@Slf4j
 public class JsonSchemaGenerator {
+    
     private static final List<Class<?>> TYPES_RESOLVED_AS_STRING = List.of(Duration.class, LocalTime.class, LocalDate.class, LocalDateTime.class, ZonedDateTime.class, OffsetDateTime.class, OffsetTime.class);
     private static final List<Class<?>> SUBTYPE_RESOLUTION_EXCLUSION_FOR_PLUGIN_SCHEMA = List.of(Task.class, AbstractTrigger.class);
 
@@ -270,8 +276,22 @@ public class JsonSchemaGenerator {
             .with(Option.DEFINITIONS_FOR_ALL_OBJECTS)
             .with(Option.DEFINITION_FOR_MAIN_SCHEMA)
             .with(Option.PLAIN_DEFINITION_KEYS)
-            .with(Option.ALLOF_CLEANUP_AT_THE_END);;
-
+            .with(Option.ALLOF_CLEANUP_AT_THE_END);
+            
+        // HACK: Registered a custom JsonUnwrappedDefinitionProvider prior to the JacksonModule 
+        // to be able to return an CustomDefinition with an empty node when the ResolvedType can't be found.
+        builder.forTypesInGeneral().withCustomDefinitionProvider(new JsonUnwrappedDefinitionProvider(){
+            @Override
+            public CustomDefinition provideCustomSchemaDefinition(ResolvedType javaType, SchemaGenerationContext context) {
+                try {
+                    return super.provideCustomSchemaDefinition(javaType, context);
+                } catch (NoClassDefFoundError e) {
+                    // This error happens when a non-supported plugin type exists in the classpath.
+                    log.debug("Cannot create schema definition for type '{}'. Cause: NoClassDefFoundError", javaType.getTypeName());
+                    return new CustomDefinition(context.getGeneratorConfig().createObjectNode(), true);
+                }
+            }
+        });
         if (!draft7) {
             builder.with(new JacksonModule(JacksonOption.IGNORE_TYPE_INFO_TRANSFORM));
         } else {
@@ -300,6 +320,7 @@ public class JsonSchemaGenerator {
         // inline some type
         builder.forTypesInGeneral()
             .withCustomDefinitionProvider(new CustomDefinitionProviderV2() {
+                
                 @Override
                 public CustomDefinition provideCustomSchemaDefinition(ResolvedType javaType, SchemaGenerationContext context) {
                     if (javaType.isInstanceOf(Map.class) || javaType.isInstanceOf(Enum.class)) {

core/src/main/java/io/kestra/core/exceptions/InvalidTriggerConfigurationException.java

@@ -0,0 +1,15 @@
+package io.kestra.core.exceptions;
+
+public class InvalidTriggerConfigurationException extends KestraRuntimeException {
+    public InvalidTriggerConfigurationException() {
+        super();
+    }
+
+    public InvalidTriggerConfigurationException(String message) {
+        super(message);
+    }
+
+    public InvalidTriggerConfigurationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

core/src/main/java/io/kestra/core/models/QueryFilter.java

@@ -250,6 +250,15 @@ public record QueryFilter(
                     Field.START_DATE, Field.END_DATE, Field.TRIGGER_ID
                 );
             }
+        },
+        SECRET_METADATA {
+            @Override
+            public List<Field> supportedField() {
+                return List.of(
+                    Field.QUERY,
+                    Field.NAMESPACE
+                );
+            }
         };
 
         public abstract List<Field> supportedField();

core/src/main/java/io/kestra/core/models/conditions/Condition.java

@@ -12,6 +12,8 @@ import lombok.experimental.SuperBuilder;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @io.kestra.core.models.annotations.Plugin
 @SuperBuilder
 @Getter
@@ -20,6 +22,6 @@ import jakarta.validation.constraints.Pattern;
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public abstract class Condition implements Plugin, Rethrow.PredicateChecked<ConditionContext, InternalException> {
     @NotNull
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 }

core/src/main/java/io/kestra/core/models/dashboards/Dashboard.java

@@ -32,6 +32,8 @@ public class Dashboard implements HasUID, DeletedInterface {
     private String tenantId;
 
     @Hidden
+    @NotNull
+    @NotBlank
     private String id;
 
     @NotNull

core/src/main/java/io/kestra/core/models/dashboards/DataFilter.java

@@ -20,6 +20,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @SuperBuilder(toBuilder = true)
 @Getter
 @NoArgsConstructor
@@ -28,7 +30,7 @@ import java.util.Set;
 public abstract class DataFilter<F extends Enum<F>, C extends ColumnDescriptor<F>> implements io.kestra.core.models.Plugin, IData<F> {
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     private String type;
 
     private Map<String, C> columns;

core/src/main/java/io/kestra/core/models/dashboards/DataFilterKPI.java

@@ -19,6 +19,8 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @SuperBuilder(toBuilder = true)
 @Getter
 @NoArgsConstructor
@@ -27,7 +29,7 @@ import java.util.Set;
 public abstract class DataFilterKPI<F extends Enum<F>, C extends ColumnDescriptor<F>> implements io.kestra.core.models.Plugin, IData<F> {
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     private String type;
 
     private C columns;

core/src/main/java/io/kestra/core/models/dashboards/charts/Chart.java

@@ -12,6 +12,8 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @SuperBuilder(toBuilder = true)
 @Getter
 @NoArgsConstructor
@@ -26,7 +28,7 @@ public abstract class Chart<P extends ChartOption> implements io.kestra.core.mod
 
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 
     @Valid

core/src/main/java/io/kestra/core/models/executions/LogEntry.java

@@ -10,6 +10,7 @@ import io.swagger.v3.oas.annotations.Hidden;
 import jakarta.annotation.Nullable;
 import lombok.Builder;
 import lombok.Value;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.event.Level;
 
 import jakarta.validation.constraints.NotNull;
@@ -120,6 +121,16 @@ public class LogEntry implements DeletedInterface, TenantInterface {
         return logEntry.getTimestamp().toString() + " " + logEntry.getLevel() + " " + logEntry.getMessage();
     }
 
+    public static String toPrettyString(LogEntry logEntry, Integer maxMessageSize) {
+        String message;
+        if (maxMessageSize != null && maxMessageSize > 0) {
+            message = StringUtils.truncate(logEntry.getMessage(), maxMessageSize);
+        } else {
+            message = logEntry.getMessage();
+        }
+        return logEntry.getTimestamp().toString() + " " + logEntry.getLevel() + " " + message;
+    }
+
     public Map<String, String> toMap() {
         return Stream
             .of(

core/src/main/java/io/kestra/core/models/executions/MetricEntry.java

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.executions.metrics.Counter;
+import io.kestra.core.models.executions.metrics.Gauge;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.swagger.v3.oas.annotations.Hidden;
 import jakarta.annotation.Nullable;
@@ -82,6 +83,10 @@ public class MetricEntry implements DeletedInterface, TenantInterface {
             return counter.getValue();
         }
 
+        if (metricEntry instanceof Gauge gauge) {
+            return gauge.getValue();
+        }
+
         if (metricEntry instanceof Timer timer) {
             return (double) timer.getValue().toMillis();
         }

core/src/main/java/io/kestra/core/models/executions/TaskRun.java

@@ -3,7 +3,9 @@ package io.kestra.core.models.executions;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.ResolvedTask;
+import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.utils.IdUtils;
 import io.swagger.v3.oas.annotations.Hidden;
@@ -52,6 +54,7 @@ public class TaskRun implements TenantInterface {
 
     @With
     @JsonInclude(JsonInclude.Include.ALWAYS)
+    @Nullable
     Variables outputs;
 
     @NotNull
@@ -64,7 +67,6 @@ public class TaskRun implements TenantInterface {
     Boolean dynamic;
 
     // Set it to true to force execution even if the execution is killed
-    @Nullable
     @With
     Boolean forceExecution;
 
@@ -217,7 +219,7 @@ public class TaskRun implements TenantInterface {
     public boolean isSame(TaskRun taskRun) {
         return this.getId().equals(taskRun.getId()) &&
             ((this.getValue() == null && taskRun.getValue() == null) || (this.getValue() != null && this.getValue().equals(taskRun.getValue()))) &&
-            ((this.getIteration() == null && taskRun.getIteration() == null) || (this.getIteration() != null && this.getIteration().equals(taskRun.getIteration()))) ;
+            ((this.getIteration() == null && taskRun.getIteration() == null) || (this.getIteration() != null && this.getIteration().equals(taskRun.getIteration())));
     }
 
     public String toString(boolean pretty) {
@@ -249,7 +251,7 @@ public class TaskRun implements TenantInterface {
      * This method is used when the retry is apply on a task
      * but the retry type is NEW_EXECUTION
      *
-     * @param retry Contains the retry configuration
+     * @param retry     Contains the retry configuration
      * @param execution Contains the attempt number and original creation date
      * @return The next retry date, null if maxAttempt || maxDuration is reached
      */
@@ -270,6 +272,7 @@ public class TaskRun implements TenantInterface {
 
     /**
      * This method is used when the Retry definition comes from the flow
+     *
      * @param retry The retry configuration
      * @return The next retry date, null if maxAttempt || maxDuration is reached
      */

core/src/main/java/io/kestra/core/models/executions/metrics/Gauge.java

@@ -0,0 +1,78 @@
+package io.kestra.core.models.executions.metrics;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import jakarta.annotation.Nullable;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.executions.AbstractMetricEntry;
+
+import jakarta.validation.constraints.NotNull;
+import java.util.Map;
+
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+public class Gauge extends AbstractMetricEntry<Double> {
+    public static final String TYPE = "gauge";
+
+    @NotNull
+    @JsonInclude
+    private final String type = TYPE;
+
+    @NotNull
+    @EqualsAndHashCode.Exclude
+    private Double value;
+
+    private Gauge(@NotNull String name, @Nullable String description, @NotNull Double value, String... tags) {
+        super(name, description, tags);
+
+        this.value = value;
+    }
+
+    public static Gauge of(@NotNull String name, @NotNull Double value, String... tags) {
+        return new Gauge(name, null, value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @Nullable String description, @NotNull Double value, String... tags) {
+        return new Gauge(name, description, value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @NotNull Integer value, String... tags) {
+        return new Gauge(name, null, (double) value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @Nullable String description, @NotNull Integer value, String... tags) {
+        return new Gauge(name, description, (double) value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @NotNull Long value, String... tags) {
+        return new Gauge(name, null, (double) value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @Nullable String description, @NotNull Long value, String... tags) {
+        return new Gauge(name, description, (double) value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @NotNull Float value, String... tags) {
+        return new Gauge(name, null, (double) value, tags);
+    }
+
+    public static Gauge of(@NotNull String name, @Nullable String description, @NotNull Float value, String... tags) {
+        return new Gauge(name, description, (double) value, tags);
+    }
+
+    @Override
+    public void register(MetricRegistry meterRegistry, String name, String description, Map<String, String> tags) {
+        meterRegistry
+                .gauge(this.metricName(name), description, this.value, this.tagsAsArray(tags));
+    }
+
+    @Override
+    public void increment(Double value) {
+        this.value = value;
+    }
+}

core/src/main/java/io/kestra/core/models/flows/AbstractFlow.java

@@ -77,14 +77,6 @@ public abstract class AbstractFlow implements FlowInterface {
     Map<String, Object> variables;
 
 
-    @Schema(
-        oneOf = {
-            String.class,  // Corresponds to 'type: string' in OpenAPI
-            Map.class      // Corresponds to 'type: object' in OpenAPI
-        }
-    )
-    interface StringOrMapValue {}
-
     @Valid
     private WorkerGroup workerGroup;
 

core/src/main/java/io/kestra/core/models/flows/Flow.java

@@ -49,7 +49,7 @@ import java.util.stream.Stream;
 public class Flow extends AbstractFlow implements HasUID {
     private static final ObjectMapper NON_DEFAULT_OBJECT_MAPPER = JacksonMapper.ofYaml()
         .copy()
-        .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
+        .setDefaultPropertyInclusion(JsonInclude.Include.NON_DEFAULT);
 
     private static final ObjectMapper WITHOUT_REVISION_OBJECT_MAPPER = NON_DEFAULT_OBJECT_MAPPER.copy()
         .configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS, true)

core/src/main/java/io/kestra/core/models/flows/FlowInterface.java

@@ -136,7 +136,7 @@ public interface FlowInterface extends FlowId, DeletedInterface, TenantInterface
     class SourceGenerator {
         private static final ObjectMapper NON_DEFAULT_OBJECT_MAPPER = JacksonMapper.ofJson()
             .copy()
-            .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
+            .setDefaultPropertyInclusion(JsonInclude.Include.NON_DEFAULT);
 
         static String generate(final FlowInterface flow) {
             try {

core/src/main/java/io/kestra/core/models/flows/PluginDefault.java

@@ -24,10 +24,6 @@ public class PluginDefault {
 
     @Schema(
         type = "object",
-        oneOf = {
-            Map.class,
-            String.class
-        },
         additionalProperties = Schema.AdditionalPropertiesValue.FALSE
     )
     private final Map<String, Object> values;

core/src/main/java/io/kestra/core/models/flows/input/FileInput.java

@@ -1,5 +1,6 @@
 package io.kestra.core.models.flows.input;
 
+import java.util.Set;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.validations.FileInputValidation;
 import jakarta.validation.ConstraintViolationException;
@@ -22,10 +23,35 @@ public class FileInput extends Input<URI> {
 
     @Deprecated(since = "0.24", forRemoval = true)
     public String extension;
+    
+    /**
+     * List of allowed file extensions (e.g., [".csv", ".txt", ".pdf"]).
+     * Each extension must start with a dot.
+     */
+    private List<String> allowedFileExtensions;
+    
+    /**
+     * Gets the file extension from the URI's path
+     */
+    private String getFileExtension(URI uri) {
+        String path = uri.getPath();
+        int lastDotIndex = path.lastIndexOf(".");
+        return lastDotIndex >= 0 ? path.substring(lastDotIndex).toLowerCase() : "";
+    }
 
     @Override
     public void validate(URI input) throws ConstraintViolationException {
-        // no validation yet
+        if (input == null || allowedFileExtensions == null || allowedFileExtensions.isEmpty()) {
+            return;
+        }
+
+        String extension = getFileExtension(input);
+        if (!allowedFileExtensions.contains(extension.toLowerCase())) {
+            throw new ConstraintViolationException(
+                "File type not allowed. Accepted extensions: " + String.join(", ", allowedFileExtensions),
+                Set.of()
+            );
+        }
     }
 
     public static String findFileInputExtension(@NotNull final List<Input<?>> inputs, @NotNull final String fileName) {

core/src/main/java/io/kestra/core/models/tasks/TaskInterface.java

@@ -8,6 +8,8 @@ import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public interface TaskInterface extends Plugin, PluginVersioning {
     @NotNull
@@ -17,7 +19,7 @@ public interface TaskInterface extends Plugin, PluginVersioning {
 
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     @Schema(title = "The class name of this task.")
     String getType();
 }

core/src/main/java/io/kestra/core/models/tasks/logs/LogExporter.java

@@ -11,6 +11,8 @@ import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 import reactor.core.publisher.Flux;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @Plugin
 @SuperBuilder(toBuilder = true)
 @Getter
@@ -22,7 +24,7 @@ public abstract class LogExporter<T extends Output>  implements io.kestra.core.m
     protected String id;
 
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 
     public abstract T sendLogs(RunContext runContext, Flux<LogRecord> logRecords) throws Exception;

core/src/main/java/io/kestra/core/models/tasks/logs/LogRecordMapper.java

@@ -8,12 +8,16 @@ public final class LogRecordMapper {
     private LogRecordMapper(){}
 
     public static LogRecord mapToLogRecord(LogEntry log) {
+        return mapToLogRecord(log, null);
+    }
+
+    public static LogRecord mapToLogRecord(LogEntry log, Integer maxMessageSize) {
         return LogRecord.builder()
             .resource("Kestra")
             .timestampEpochNanos(instantInNanos(log.getTimestamp()))
             .severity(log.getLevel().name())
             .attributes(log.toLogMap())
-            .bodyValue(LogEntry.toPrettyString(log))
+            .bodyValue(LogEntry.toPrettyString(log, maxMessageSize))
             .build();
     }
 

core/src/main/java/io/kestra/core/models/tasks/metrics/AbstractMetric.java

@@ -22,6 +22,7 @@ import java.util.Map;
 @JsonSubTypes({
     @JsonSubTypes.Type(value = CounterMetric.class, name = "counter"),
     @JsonSubTypes.Type(value = TimerMetric.class, name = "timer"),
+    @JsonSubTypes.Type(value = GaugeMetric.class, name = "gauge"),
 })
 @ToString
 @EqualsAndHashCode

core/src/main/java/io/kestra/core/models/tasks/metrics/GaugeMetric.java

@@ -0,0 +1,44 @@
+package io.kestra.core.models.tasks.metrics;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.executions.AbstractMetricEntry;
+import io.kestra.core.models.executions.metrics.Gauge;
+import io.kestra.core.models.property.Property;
+import io.kestra.core.runners.RunContext;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public class GaugeMetric extends AbstractMetric {
+    public static final String TYPE = "gauge";
+
+    @NotNull
+    @EqualsAndHashCode.Exclude
+    private Property<Double> value;
+
+    @Override
+    public AbstractMetricEntry<?> toMetric(RunContext runContext) throws IllegalVariableEvaluationException {
+        String name = runContext.render(this.name).as(String.class).orElseThrow();
+        Double value = runContext.render(this.value).as(Double.class).orElseThrow();
+        String description = runContext.render(this.description).as(String.class).orElse(null);
+        Map<String, String> tags = runContext.render(this.tags).asMap(String.class, String.class);
+        String[] tagsAsStrings = tags.entrySet().stream()
+                .flatMap(e -> Stream.of(e.getKey(), e.getValue()))
+                .toArray(String[]::new);
+
+        return Gauge.of(name, description, value, tagsAsStrings);
+    }
+
+    public String getType() {
+        return TYPE;
+    }
+}

core/src/main/java/io/kestra/core/models/tasks/runners/TaskRunner.java

@@ -25,6 +25,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static io.kestra.core.utils.WindowsUtils.windowsToUnixPath;
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
 
 /**
  * Base class for all task runners.
@@ -36,7 +37,7 @@ import static io.kestra.core.utils.WindowsUtils.windowsToUnixPath;
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public abstract class TaskRunner<T extends TaskRunnerDetailResult> implements Plugin, PluginVersioning, WorkerJobLifecycle {
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 
     @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)

core/src/main/java/io/kestra/core/models/templates/Template.java

@@ -44,7 +44,7 @@ public class Template implements DeletedInterface, TenantInterface, HasUID {
                 return exclusions.contains(m.getName()) || super.hasIgnoreMarker(m);
             }
         })
-        .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
+        .setDefaultPropertyInclusion(JsonInclude.Include.NON_DEFAULT);
 
     @Setter
     @Hidden

core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java

@@ -47,7 +47,6 @@ abstract public class AbstractTrigger implements TriggerInterface {
     @Valid
     protected List<@Valid @NotNull Condition> conditions;
 
-    @NotNull
     @Builder.Default
     @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
     @Schema(defaultValue = "false")

core/src/main/java/io/kestra/core/models/triggers/PollingTriggerInterface.java

@@ -1,10 +1,12 @@
 package io.kestra.core.models.triggers;
 
+import io.kestra.core.exceptions.InvalidTriggerConfigurationException;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.swagger.v3.oas.annotations.media.Schema;
 
+import java.time.DateTimeException;
 import java.time.Duration;
 import java.time.ZonedDateTime;
 import java.util.Optional;
@@ -29,15 +31,29 @@ public interface PollingTriggerInterface extends WorkerTriggerInterface {
      * Compute the next evaluation date of the trigger based on the existing trigger context: by default, it uses the current date and the interval.
      * Schedulable triggers must override this method.
      */
-    default ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws Exception {
-        return ZonedDateTime.now().plus(this.getInterval());
+    default ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws InvalidTriggerConfigurationException {
+        return computeNextEvaluationDate();
     }
 
     /**
      * Compute the next evaluation date of the trigger: by default, it uses the current date and the interval.
      * Schedulable triggers must override this method as it's used to init them when there is no evaluation date.
      */
-    default ZonedDateTime nextEvaluationDate() {
-        return ZonedDateTime.now().plus(this.getInterval());
+    default ZonedDateTime nextEvaluationDate() throws InvalidTriggerConfigurationException {
+        return computeNextEvaluationDate();
+    }
+
+    /**
+     * computes the next evaluation date using the configured interval.
+     * Throw InvalidTriggerConfigurationException, if the interval causes date overflow.
+     */
+    private ZonedDateTime computeNextEvaluationDate() throws InvalidTriggerConfigurationException {
+        Duration interval = this.getInterval();
+
+        try {
+            return ZonedDateTime.now().plus(interval);
+        } catch (DateTimeException | ArithmeticException e) {
+            throw new InvalidTriggerConfigurationException("Trigger interval too large", e);
+        }
     }
 }

core/src/main/java/io/kestra/core/models/triggers/StatefulTriggerInterface.java

@@ -0,0 +1,40 @@
+package io.kestra.core.models.triggers;
+
+import io.kestra.core.models.property.Property;
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import java.time.Duration;
+
+public interface StatefulTriggerInterface {
+    @Schema(
+        title = "Trigger event type",
+        description = """
+            Defines when the trigger fires.
+            - `CREATE`: only for newly discovered entities.
+            - `UPDATE`: only when an already-seen entity changes.
+            - `CREATE_OR_UPDATE`: fires on either event.
+            """
+    )
+    Property<On> getOn();
+
+    @Schema(
+        title = "State key",
+        description = """
+            JSON-type KV key for persisted state.
+            Default: `<namespace>__<flowId>__<triggerId>`
+            """
+    )
+    Property<String> getStateKey();
+
+    @Schema(
+        title = "State TTL",
+        description = "TTL for persisted state entries (e.g., PT24H, P7D)."
+    )
+    Property<Duration> getStateTtl();
+
+    enum On {
+        CREATE,
+        UPDATE,
+        CREATE_OR_UPDATE
+    }
+}

core/src/main/java/io/kestra/core/models/triggers/StatefulTriggerService.java

@@ -0,0 +1,91 @@
+package io.kestra.core.models.triggers;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.storages.kv.KVMetadata;
+import io.kestra.core.storages.kv.KVValueAndMetadata;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class StatefulTriggerService {
+    public record Entry(String uri, String version, Instant modifiedAt, Instant lastSeenAt) {
+        public static Entry candidate(String uri, String version, Instant modifiedAt) {
+            return new Entry(uri, version, modifiedAt, null);
+        }
+    }
+
+    public record StateUpdate(boolean fire, boolean isNew) {}
+
+    public static Map<String, Entry> readState(RunContext runContext, String key, Optional<Duration> ttl) {
+        try {
+            var kv = runContext.namespaceKv(runContext.flowInfo().namespace()).getValue(key);
+            if (kv.isEmpty()) {
+                return new HashMap<>();
+            }
+
+            var entries = JacksonMapper.ofJson().readValue((byte[]) kv.get().value(), new TypeReference<List<Entry>>() {});
+
+            var cutoff = ttl.map(d -> Instant.now().minus(d)).orElse(Instant.MIN);
+
+            return entries.stream()
+                .filter(e -> Optional.ofNullable(e.lastSeenAt()).orElse(Instant.now()).isAfter(cutoff))
+                .collect(Collectors.toMap(Entry::uri, e -> e));
+        } catch (Exception e) {
+            runContext.logger().warn("readState failed", e);
+            return new HashMap<>();
+        }
+    }
+
+    public static void writeState(RunContext runContext, String key, Map<String, Entry> state, Optional<Duration> ttl) {
+        try {
+            var bytes = JacksonMapper.ofJson().writeValueAsBytes(state.values());
+            var meta = new KVMetadata("trigger state", ttl.orElse(null));
+
+            runContext.namespaceKv(runContext.flowInfo().namespace()).put(key, new KVValueAndMetadata(meta, bytes));
+        } catch (Exception e) {
+            runContext.logger().warn("writeState failed", e);
+        }
+    }
+
+    public static StateUpdate computeAndUpdateState(Map<String, Entry> state, Entry candidate, StatefulTriggerInterface.On on) {
+        var prev = state.get(candidate.uri());
+        var isNew = prev == null;
+        var fire = shouldFire(prev, candidate.version(), on);
+
+        Instant lastSeenAt;
+
+        if (fire || isNew) {
+            // it is new seen or changed
+            lastSeenAt = Instant.now();
+        } else if (prev.lastSeenAt() != null) {
+            // it is unchanged but already tracked before
+            lastSeenAt = prev.lastSeenAt();
+        } else {
+            lastSeenAt = Instant.now();
+        }
+
+        var newEntry = new Entry(candidate.uri(), candidate.version(), candidate.modifiedAt(), lastSeenAt);
+
+        state.put(candidate.uri(), newEntry);
+
+        return new StatefulTriggerService.StateUpdate(fire, isNew);
+    }
+
+    public static boolean shouldFire(Entry prev, String version, StatefulTriggerInterface.On on) {
+        if (prev == null) {
+            return on == StatefulTriggerInterface.On.CREATE || on == StatefulTriggerInterface.On.CREATE_OR_UPDATE;
+        }
+        if (!Objects.equals(prev.version(), version)) {
+            return on == StatefulTriggerInterface.On.UPDATE || on == StatefulTriggerInterface.On.CREATE_OR_UPDATE;
+        }
+        return false;
+    }
+
+    public static String defaultKey(String ns, String flowId, String triggerId) {
+        return String.join("_", ns, flowId, triggerId);
+    }
+}

core/src/main/java/io/kestra/core/models/triggers/Trigger.java

@@ -1,5 +1,6 @@
 package io.kestra.core.models.triggers;
 
+import io.kestra.core.exceptions.InvalidTriggerConfigurationException;
 import io.kestra.core.models.HasUID;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
@@ -167,9 +168,14 @@ public class Trigger extends TriggerContext implements HasUID {
     // Used to update trigger in flowListeners
     public static Trigger of(FlowInterface flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext, Optional<Trigger> lastTrigger) throws Exception {
         ZonedDateTime nextDate = null;
+        boolean disabled = lastTrigger.map(TriggerContext::getDisabled).orElse(Boolean.FALSE);
 
         if (abstractTrigger instanceof PollingTriggerInterface pollingTriggerInterface) {
-            nextDate = pollingTriggerInterface.nextEvaluationDate(conditionContext, Optional.empty());
+            try {
+                nextDate = pollingTriggerInterface.nextEvaluationDate(conditionContext, Optional.empty());
+            } catch (InvalidTriggerConfigurationException e) {
+                disabled = true;
+            }
         }
 
         return Trigger.builder()
@@ -180,7 +186,7 @@ public class Trigger extends TriggerContext implements HasUID {
             .date(ZonedDateTime.now().truncatedTo(ChronoUnit.SECONDS))
             .nextExecutionDate(nextDate)
             .stopAfter(abstractTrigger.getStopAfter())
-            .disabled(lastTrigger.map(TriggerContext::getDisabled).orElse(Boolean.FALSE))
+            .disabled(disabled)
             .backfill(null)
             .build();
     }

core/src/main/java/io/kestra/core/models/triggers/TriggerInterface.java

@@ -7,6 +7,7 @@ import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
 
 public interface TriggerInterface extends Plugin, PluginVersioning {
     @NotNull
@@ -17,7 +18,7 @@ public interface TriggerInterface extends Plugin, PluginVersioning {
 
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     @Schema(title = "The class name for this current trigger.")
     String getType();
 

core/src/main/java/io/kestra/core/models/triggers/TriggerService.java

@@ -11,7 +11,6 @@ import io.kestra.core.runners.FlowInputOutput;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.ListUtils;
-
 import java.time.ZonedDateTime;
 import java.util.*;
 
@@ -25,7 +24,7 @@ public abstract class TriggerService {
         RunContext runContext = conditionContext.getRunContext();
         ExecutionTrigger executionTrigger = ExecutionTrigger.of(trigger, variables, runContext.logFileURI());
 
-        return generateExecution(runContext.getTriggerExecutionId(), trigger, context, executionTrigger, conditionContext.getFlow().getRevision());
+        return generateExecution(runContext.getTriggerExecutionId(), trigger, context, executionTrigger, conditionContext);
     }
 
     public static Execution generateExecution(
@@ -37,7 +36,7 @@ public abstract class TriggerService {
         RunContext runContext = conditionContext.getRunContext();
         ExecutionTrigger executionTrigger = ExecutionTrigger.of(trigger, output, runContext.logFileURI());
 
-        return generateExecution(runContext.getTriggerExecutionId(), trigger, context, executionTrigger, conditionContext.getFlow().getRevision());
+        return generateExecution(runContext.getTriggerExecutionId(), trigger, context, executionTrigger, conditionContext);
     }
 
     public static Execution generateRealtimeExecution(
@@ -49,7 +48,7 @@ public abstract class TriggerService {
         RunContext runContext = conditionContext.getRunContext();
         ExecutionTrigger executionTrigger = ExecutionTrigger.of(trigger, output, runContext.logFileURI());
 
-        return generateExecution(IdUtils.create(), trigger, context, executionTrigger, conditionContext.getFlow().getRevision());
+        return generateExecution(IdUtils.create(), trigger, context, executionTrigger, conditionContext);
     }
 
     public static Execution generateScheduledExecution(
@@ -75,6 +74,7 @@ public abstract class TriggerService {
             .namespace(context.getNamespace())
             .flowId(context.getFlowId())
             .flowRevision(conditionContext.getFlow().getRevision())
+            .variables(conditionContext.getFlow().getVariables())
             .labels(executionLabels)
             .state(new State())
             .trigger(executionTrigger)
@@ -108,7 +108,7 @@ public abstract class TriggerService {
         AbstractTrigger trigger,
         TriggerContext context,
         ExecutionTrigger executionTrigger,
-        Integer flowRevision
+        ConditionContext conditionContext
     ) {
         List<Label> executionLabels = new ArrayList<>(ListUtils.emptyOnNull(trigger.getLabels()));
         if (executionLabels.stream().noneMatch(label -> Label.CORRELATION_ID.equals(label.key()))) {
@@ -120,7 +120,8 @@ public abstract class TriggerService {
             .namespace(context.getNamespace())
             .flowId(context.getFlowId())
             .tenantId(context.getTenantId())
-            .flowRevision(flowRevision)
+            .flowRevision(conditionContext.getFlow().getRevision())
+            .variables(conditionContext.getFlow().getVariables())
             .state(new State())
             .trigger(executionTrigger)
             .labels(executionLabels)

core/src/main/java/io/kestra/core/plugins/AdditionalPlugin.java

@@ -8,6 +8,8 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @io.kestra.core.models.annotations.Plugin
 @SuperBuilder(toBuilder = true)
 @Getter
@@ -15,6 +17,6 @@ import lombok.experimental.SuperBuilder;
 public abstract class AdditionalPlugin implements Plugin {
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 }

core/src/main/java/io/kestra/core/runners/FlowInputOutput.java

@@ -287,9 +287,10 @@ public class FlowInputOutput {
         Input<?> input = resolvable.get().input();
 
         try {
-            //  resolve all input dependencies and check whether input is enabled
-            final Map<String, InputAndValue> dependencies = resolveAllDependentInputs(input, flow, execution, inputs, decryptSecrets);
-            final RunContext runContext = buildRunContextForExecutionAndInputs(flow, execution, dependencies, decryptSecrets);
+            // Resolve all input dependencies and check whether input is enabled
+            // Note: Secrets are always decrypted here because they can be part of expressions used to render inputs such as SELECT & MULTI_SELECT.
+            final Map<String, InputAndValue> dependencies = resolveAllDependentInputs(input, flow, execution, inputs, true);
+            final RunContext runContext = buildRunContextForExecutionAndInputs(flow, execution, dependencies, true);
 
             boolean isInputEnabled = dependencies.isEmpty() || dependencies.values().stream().allMatch(InputAndValue::enabled);
 
@@ -329,7 +330,8 @@ public class FlowInputOutput {
 
             // resolve default if needed
             if (value == null && input.getDefaults() != null) {
-                value = resolveDefaultValue(input, runContext);
+                RunContext runContextForDefault = decryptSecrets ? runContext : buildRunContextForExecutionAndInputs(flow, execution, dependencies, false);
+                value = resolveDefaultValue(input, runContextForDefault);
                 resolvable.isDefault(true);
             }
 

core/src/main/java/io/kestra/core/runners/pebble/functions/KvFunction.java

@@ -38,7 +38,7 @@ public class KvFunction implements Function {
         String key = getKey(args, self, lineNumber);
         String namespace = (String) args.get(NAMESPACE_ARG);
 
-        Boolean errorOnMissing = Optional.ofNullable((Boolean) args.get(ERROR_ON_MISSING_ARG)).orElse(true);
+        boolean errorOnMissing = Optional.ofNullable((Boolean) args.get(ERROR_ON_MISSING_ARG)).orElse(true);
 
         Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
         String flowNamespace = flow.get(NAMESPACE_ARG);
@@ -53,11 +53,16 @@ public class KvFunction implements Function {
                 // we didn't check allowedNamespace here as it's checked in the kvStoreService itself
                 value = kvStoreService.get(flowTenantId, namespace, flowNamespace).getValue(key);
             }
+        } catch (ResourceExpiredException e) {
+            if (errorOnMissing) {
+                throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
+            }
+            value = Optional.empty();
         } catch (Exception e) {
             throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
         }
 
-        if (value.isEmpty() && errorOnMissing == Boolean.TRUE) {
+        if (value.isEmpty() && errorOnMissing) {
             throw new PebbleException(null, "The key '" + key + "' does not exist in the namespace '" + namespace + "'.", lineNumber, self.getName());
         }
 
@@ -85,4 +90,4 @@ public class KvFunction implements Function {
 
         return (String) args.get(KEY_ARGS);
     }
-}
+}

core/src/main/java/io/kestra/core/secret/SecretService.java

@@ -1,21 +1,29 @@
 package io.kestra.core.secret;
 
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.repositories.ArrayListTotal;
+import io.micronaut.data.model.Pageable;
 import jakarta.annotation.PostConstruct;
 import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.Strings;
 
 import java.io.IOException;
 import java.util.Base64;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 @Singleton
 @Slf4j
-public class SecretService {
+public class SecretService<META> {
     private static final String SECRET_PREFIX = "SECRET_";
+
     private Map<String, String> decodedSecrets;
 
+
     @PostConstruct
     private void postConstruct() {
         this.decode();
@@ -46,6 +54,28 @@ public class SecretService {
         return secret;
     }
 
+    public ArrayListTotal<META> list(Pageable pageable, String tenantId, List<QueryFilter> filters) throws IOException {
+        final Predicate<String> queryPredicate = filters.stream()
+            .filter(filter -> filter.field().equals(QueryFilter.Field.QUERY) && filter.value() != null)
+            .findFirst()
+            .map(filter -> {
+                if (filter.operation().equals(QueryFilter.Op.EQUALS)) {
+                    return (Predicate<String>) s -> Strings.CI.contains(s, (String) filter.value());
+                } else if (filter.operation().equals(QueryFilter.Op.NOT_EQUALS)) {
+                    return (Predicate<String>) s -> !Strings.CI.contains(s, (String) filter.value());
+                } else {
+                    throw new IllegalArgumentException("Unsupported operation for QUERY filter: " + filter.operation());
+                }
+            })
+            .orElse(s -> true);
+
+        //noinspection unchecked
+        return ArrayListTotal.of(
+            pageable,
+            decodedSecrets.keySet().stream().filter(queryPredicate).map(s -> (META) s).toList()
+        );
+    }
+
     public Map<String, Set<String>> inheritedSecrets(String tenantId, String namespace) throws IOException {
         return Map.of(namespace, decodedSecrets.keySet());
     }

core/src/main/java/io/kestra/core/serializers/JacksonMapper.java

@@ -140,7 +140,7 @@ public final class JacksonMapper {
 
         return mapper
             .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
-            .setSerializationInclusion(JsonInclude.Include.NON_NULL)
+            .setDefaultPropertyInclusion(JsonInclude.Include.NON_NULL)
             .registerModule(new JavaTimeModule())
             .registerModule(new Jdk8Module())
             .registerModule(new ParameterNamesModule())
@@ -153,7 +153,7 @@ public final class JacksonMapper {
 
     private static ObjectMapper createIonObjectMapper() {
         return configure(new IonObjectMapper(new IonFactory(createIonSystem())))
-            .setSerializationInclusion(JsonInclude.Include.ALWAYS)
+            .setDefaultPropertyInclusion(JsonInclude.Include.ALWAYS)
             .registerModule(new IonModule());
     }
 
@@ -172,7 +172,7 @@ public final class JacksonMapper {
 
         return Pair.of(patch, revert);
     }
-    
+
     public static JsonNode applyPatchesOnJsonNode(JsonNode jsonObject, List<JsonNode> patches) {
         for (JsonNode patch : patches) {
             try {

core/src/main/java/io/kestra/core/services/PluginDefaultService.java

@@ -58,10 +58,10 @@ import java.util.stream.Stream;
 public class PluginDefaultService {
     private static final ObjectMapper NON_DEFAULT_OBJECT_MAPPER = JacksonMapper.ofYaml()
         .copy()
-        .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
+        .setDefaultPropertyInclusion(JsonInclude.Include.NON_DEFAULT);
 
     private static final ObjectMapper OBJECT_MAPPER = JacksonMapper.ofYaml().copy()
-        .setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        .setDefaultPropertyInclusion(JsonInclude.Include.NON_NULL);
     private static final String PLUGIN_DEFAULTS_FIELD = "pluginDefaults";
 
     private static final TypeReference<List<PluginDefault>> PLUGIN_DEFAULTS_TYPE_REF = new TypeReference<>() {

core/src/main/java/io/kestra/core/storages/StorageContext.java

@@ -4,6 +4,7 @@ import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.FlowId;
 import io.kestra.core.utils.Hashing;
 import io.kestra.core.utils.Slugify;
 import jakarta.annotation.Nullable;
@@ -61,11 +62,11 @@ public class StorageContext {
             taskRun.getValue()
         );
     }
-
+    
     /**
      * Factory method for constructing a new {@link StorageContext} scoped to a given {@link Flow}.
      */
-    public static StorageContext forFlow(Flow flow) {
+    public static StorageContext forFlow(FlowId flow) {
         return new StorageContext(flow.getTenantId(), flow.getNamespace(), flow.getId());
     }
 

core/src/main/java/io/kestra/core/utils/RegexPatterns.java

@@ -0,0 +1,5 @@
+package io.kestra.core.utils;
+
+public class RegexPatterns {
+    public static final String JAVA_IDENTIFIER_REGEX = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$";
+}

core/src/main/java/io/kestra/plugin/core/condition/Expression.java

@@ -40,14 +40,14 @@ import jakarta.validation.constraints.NotNull;
                   - id: hello
                     type: io.kestra.plugin.core.log.Log
                     message: Average value has gone below 10
-                
+
                 triggers:
                   - id: expression_trigger
                     type: io.kestra.plugin.core.trigger.Schedule
                     cron: "*/1 * * * *"
                     conditions:
                       - type: io.kestra.plugin.core.condition.Expression
-                        expression: "{{ kv('average_value') < 10 }}"    
+                        expression: "{{ kv('average_value') < 10 }}"
                 """
         )
     },

core/src/main/java/io/kestra/plugin/core/condition/PublicHoliday.java

@@ -16,6 +16,7 @@ import lombok.*;
 import lombok.experimental.SuperBuilder;
 
 import java.time.LocalDate;
+import java.util.Map;
 
 @SuperBuilder
 @ToString
@@ -82,7 +83,7 @@ public class PublicHoliday extends Condition implements ScheduleCondition {
     )
     @NotNull
     @Builder.Default
-    private Property<String> date = Property.ofExpression("{{ trigger.date }}");
+    private Property<String> date = Property.ofExpression("{{ trigger.date}}");
 
     @Schema(
         title = "[ISO 3166-1 alpha-2](https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2) country code. If not set, it uses the country code from the default locale.",
@@ -98,11 +99,12 @@ public class PublicHoliday extends Condition implements ScheduleCondition {
 
     @Override
     public boolean test(ConditionContext conditionContext) throws InternalException {
+        Map<String, Object> variables=conditionContext.getVariables();
         var renderedCountry = conditionContext.getRunContext().render(this.country).as(String.class).orElse(null);
         var renderedSubDivision = conditionContext.getRunContext().render(this.subDivision).as(String.class).orElse(null);
 
         HolidayManager holidayManager = renderedCountry != null ? HolidayManager.getInstance(ManagerParameters.create(renderedCountry)) : HolidayManager.getInstance();
-        LocalDate currentDate = DateUtils.parseLocalDate(conditionContext.getRunContext().render(date).as(String.class).orElseThrow());
+        LocalDate currentDate = DateUtils.parseLocalDate(conditionContext.getRunContext().render(date).as(String.class,variables).orElseThrow());
         return renderedSubDivision == null ? holidayManager.isHoliday(currentDate) : holidayManager.isHoliday(currentDate, renderedSubDivision);
     }
 }

core/src/main/java/io/kestra/plugin/core/dashboard/chart/Markdown.java

@@ -35,7 +35,7 @@ import lombok.experimental.SuperBuilder;
                     content: |
                       ## Execution Success Rate
                       This chart displays the percentage of successful executions over time.
-                    
+
                       - A **higher success rate** indicates stable and reliable workflows.
 
                       - Sudden **drops** may signal issues in task execution or external dependencies.

core/src/main/java/io/kestra/plugin/core/dashboard/chart/TimeSeries.java

@@ -57,7 +57,7 @@ import lombok.experimental.SuperBuilder;
                                 field: DURATION
                                 agg: SUM
                                 graphStyle: LINES
-                    """
+                """
             }
         )
     }

core/src/main/java/io/kestra/plugin/core/dashboard/chart/mardown/sources/MarkdownSource.java

@@ -7,6 +7,8 @@ import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import lombok.Getter;
 
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
+
 @Getter
 @JsonTypeInfo(
     use = JsonTypeInfo.Id.NAME,
@@ -20,6 +22,6 @@ import lombok.Getter;
 public class MarkdownSource {
     @NotNull
     @NotBlank
-    @Pattern(regexp = "^[A-Za-z_$][A-Za-z0-9_$]*(\\.[A-Za-z_$][A-Za-z0-9_$]*)*$")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     private String type;
 }

core/src/main/java/io/kestra/plugin/core/dashboard/data/Executions.java

@@ -51,7 +51,7 @@ import lombok.experimental.SuperBuilder;
                       displayName: Executions
                       agg: COUNT
             """
-          }                    
+          }
         )
     }
 )

core/src/main/java/io/kestra/plugin/core/debug/Return.java

@@ -24,8 +24,10 @@ import java.util.Optional;
 @NoArgsConstructor
 @Schema(
     title = "Return a value for debugging purposes.",
-    description = "This task is mostly useful for troubleshooting.\n\n" +
-        "It allows you to return some templated functions, inputs or outputs. In some cases you might want to trim all white spaces from the rendered values so downstream tasks can use them properly"
+    description = """
+        This task is mostly useful for troubleshooting.
+
+        It allows you to return some templated functions, inputs or outputs. In some cases you might want to trim all white spaces from the rendered values so downstream tasks can use them properly."""
 )
 @Plugin(
     examples = {

core/src/main/java/io/kestra/plugin/core/trigger/ScheduleOnDates.java

@@ -1,6 +1,7 @@
 package io.kestra.plugin.core.trigger;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.exceptions.InvalidTriggerConfigurationException;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.ConditionContext;
@@ -96,16 +97,29 @@ public class ScheduleOnDates extends AbstractTrigger implements Schedulable, Tri
     }
 
     @Override
-    public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws Exception {
-        // lastEvaluation date is the last one from the trigger context or the first date of the list
-        return last
-            .map(throwFunction(context -> nextDate(conditionContext.getRunContext(), date -> date.isAfter(context.getDate()))
-                .orElse(ZonedDateTime.now().plusYears(1) // it's not ideal, but we need a date or the trigger will keep evaluated
-            )))
-            .orElse(conditionContext.getRunContext().render(dates).asList(ZonedDateTime.class).stream().sorted().findFirst().orElse(ZonedDateTime.now()))
-            .truncatedTo(ChronoUnit.SECONDS);
+    public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) {
+        try {
+            return last
+                .map(throwFunction(context ->
+                    nextDate(conditionContext.getRunContext(), date -> date.isAfter(context.getDate()))
+                        .orElse(ZonedDateTime.now().plusYears(1))
+                ))
+                .orElse(conditionContext.getRunContext()
+                    .render(dates)
+                    .asList(ZonedDateTime.class)
+                    .stream()
+                    .sorted()
+                    .findFirst()
+                    .orElse(ZonedDateTime.now()))
+                .truncatedTo(ChronoUnit.SECONDS);
+        } catch (IllegalVariableEvaluationException e) {
+            log.warn("Failed to evaluate schedule dates for trigger '{}': {}", this.getId(), e.getMessage());
+            return ZonedDateTime.now().plusYears(1);
+        }
     }
 
+
+
     @Override
     public ZonedDateTime nextEvaluationDate() {
         // TODO this may be the next date from now?

core/src/main/java/io/kestra/plugin/core/trigger/Webhook.java

@@ -131,7 +131,7 @@ import jakarta.validation.constraints.Size;
 @WebhookValidation
 public class Webhook extends AbstractTrigger implements TriggerOutput<Webhook.Output> {
     private static final ObjectMapper MAPPER = JacksonMapper.ofJson().copy()
-        .setSerializationInclusion(JsonInclude.Include.USE_DEFAULTS);
+        .setDefaultPropertyInclusion(JsonInclude.Include.USE_DEFAULTS);
 
     @Size(max = 256)
     @NotNull
@@ -156,6 +156,13 @@ public class Webhook extends AbstractTrigger implements TriggerOutput<Webhook.Ou
            """
     )
     private Boolean wait = false;
+    
+    
+    @Schema(
+        title = "The inputs to pass to the triggered flow"
+    )
+    @PluginProperty(dynamic = true)
+    private Map<String, Object> inputs;
 
     @PluginProperty
     @Builder.Default
@@ -174,6 +181,7 @@ public class Webhook extends AbstractTrigger implements TriggerOutput<Webhook.Ou
             .namespace(flow.getNamespace())
             .flowId(flow.getId())
             .flowRevision(flow.getRevision())
+            .inputs(inputs)
             .state(new State())
             .trigger(ExecutionTrigger.of(
                 this,

core/src/test/java/io/kestra/core/models/flows/input/FileInputTest.java

@@ -1,10 +1,18 @@
 package io.kestra.core.models.flows.input;
 
 import io.kestra.core.models.flows.Input;
+import jakarta.validation.ConstraintViolationException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.net.URI;
 import java.util.List;
+import java.util.Arrays;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.jupiter.api.Assertions.*;
+
 class FileInputTest {
 
     @Test
@@ -28,4 +36,101 @@ class FileInputTest {
         String result = FileInput.findFileInputExtension(inputs, "???");
         Assertions.assertEquals(".upl", result);
     }
-}
+
+    @Test
+    void validateValidFileTypes() {
+        final FileInput csvInput = FileInput.builder()
+            .id("csvFile")
+            .allowedFileExtensions(List.of(".csv"))
+            .build();
+
+        // Test valid CSV file
+        assertDoesNotThrow(() -> csvInput.validate(URI.create("file:///path/to/file.csv")));
+        assertDoesNotThrow(() -> csvInput.validate(URI.create("nsfile:///path/to/file.CSV"))); // Test case-insensitive
+
+        // Test multiple extensions
+        final FileInput docInput = FileInput.builder()
+            .id("docFile")
+            .allowedFileExtensions(List.of(".doc", ".docx", ".pdf"))
+            .build();
+
+        assertDoesNotThrow(() -> docInput.validate(URI.create("file:///path/to/file.doc")));
+        assertDoesNotThrow(() -> docInput.validate(URI.create("file:///path/to/file.docx")));
+        assertDoesNotThrow(() -> docInput.validate(URI.create("file:///path/to/file.pdf")));
+    }
+
+    @Test
+    void validateInvalidFileTypes() {
+        final FileInput csvInput = FileInput.builder()
+            .id("csvFile")
+            .allowedFileExtensions(List.of(".csv"))
+            .build();
+
+        // Test invalid extension
+        ConstraintViolationException exception = assertThrows(
+            ConstraintViolationException.class,
+            () -> csvInput.validate(URI.create("file:///path/to/file.txt"))
+        );
+        assertThat(exception.getMessage(), containsString("Accepted extensions: .csv"));
+
+        // Test multiple allowed types
+        final FileInput imageInput = FileInput.builder()
+            .id("imageFile")
+            .allowedFileExtensions(List.of(".jpg", ".png"))
+            .build();
+
+        exception = assertThrows(
+            ConstraintViolationException.class,
+            () -> imageInput.validate(URI.create("file:///path/to/file.gif"))
+        );
+        assertThat(exception.getMessage(), containsString("Accepted extensions: .jpg, .png"));
+    }
+
+    @Test
+    void validateMimeTypes() {
+        final FileInput textInput = FileInput.builder()
+            .id("textFile")
+            .allowedFileExtensions(List.of(".csv", ".json"))
+            .build();
+
+        // Test valid file types
+        assertDoesNotThrow(() -> textInput.validate(URI.create("file:///path/to/file.csv")));
+        assertDoesNotThrow(() -> textInput.validate(URI.create("file:///path/to/file.json")));
+
+        // Test invalid file type
+        ConstraintViolationException exception = assertThrows(
+            ConstraintViolationException.class,
+            () -> textInput.validate(URI.create("file:///path/to/file.xml"))
+        );
+        assertThat(exception.getMessage(), containsString("Accepted extensions: .csv, .json"));
+    }
+
+    @Test
+    void validateNullValues() {
+        final FileInput csvInput = FileInput.builder()
+            .id("csvFile")
+            .allowedFileExtensions(List.of(".csv"))
+            .build();
+
+        // Null input should be allowed (for optional inputs)
+        assertDoesNotThrow(() -> csvInput.validate(null));
+
+        // Null extensions should not enforce any validation
+        final FileInput anyInput = FileInput.builder()
+            .id("anyFile")
+            .allowedFileExtensions(null)
+            .build();
+        assertDoesNotThrow(() -> anyInput.validate(URI.create("file:///path/to/any.file")));
+    }
+
+    @Test
+    void validateEmptyAccept() {
+        final FileInput anyInput = FileInput.builder()
+            .id("anyFile")
+            .allowedFileExtensions(List.of())
+            .build();
+
+        // Empty extensions list should not enforce any validation
+        assertDoesNotThrow(() -> anyInput.validate(URI.create("file:///path/to/any.file")));
+    }
+}

core/src/test/java/io/kestra/core/models/tasks/logs/LogRecordMapperTest.java

@@ -45,6 +45,29 @@ public class LogRecordMapperTest {
         softly.then(logRecord.getBodyValue()).isEqualTo("2011-12-03T10:15:30.123456789Z INFO message");
     }
 
+    @Test
+    public void should_map_with_truncate(){
+        LogEntry logEntry = LogEntry.builder()
+            .tenantId("tenantId")
+            .namespace("namespace")
+            .flowId("flowId")
+            .taskId("taskId")
+            .executionId("executionId")
+            .taskRunId("taskRunId")
+            .attemptNumber(1)
+            .triggerId("triggerId")
+            .timestamp(Instant.parse("2011-12-03T10:15:30.123456789Z"))
+            .level(Level.INFO)
+            .thread("thread")
+            .message("message")
+            .build();
+        LogRecord logRecord = LogRecordMapper.mapToLogRecord(logEntry, 1);
+        assertThat(logRecord.getBodyValue()).isEqualTo("2011-12-03T10:15:30.123456789Z INFO m");
+
+        logRecord = LogRecordMapper.mapToLogRecord(logEntry, 0);
+        assertThat(logRecord.getBodyValue()).isEqualTo("2011-12-03T10:15:30.123456789Z INFO message");
+    }
+
     @Test
     public void should_convert_instant_in_nanos(){
         Instant instant = Instant.parse("2011-12-03T10:15:30.123456789Z");

core/src/test/java/io/kestra/core/models/triggers/StatefulTriggerInterfaceTest.java

@@ -0,0 +1,91 @@
+package io.kestra.core.models.triggers;
+
+import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.runners.RunContextFactory;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.kestra.core.models.triggers.StatefulTriggerService.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+@KestraTest
+class StatefulTriggerInterfaceTest {
+    @Inject
+    RunContextFactory runContextFactory;
+    
+    @Test
+    void shouldPersistAndReadState() throws Exception {
+        var flow = Flow.builder()
+            .namespace("io.kestra.unittest")
+            .id("test-flow")
+            .revision(1)
+            .build();
+
+        var runContext = runContextFactory.of(flow, Map.of(
+            "flow", Map.of(
+                "tenantId", "main",
+                "namespace", "io.kestra.unittest",
+                "id", "test-flow",
+                "revision", 1
+            )
+        ));
+
+        var key = defaultKey("ns", "test-flow", "trigger-persist");
+        var ttl = Optional.of(Duration.ofMinutes(5));
+        var state = new HashMap<String, StatefulTriggerService.Entry>();
+
+        var candidate = StatefulTriggerService.Entry.candidate("gs://bucket/file1.csv", "v1", Instant.now());
+        var result = computeAndUpdateState(state, candidate, StatefulTriggerInterface.On.CREATE_OR_UPDATE);
+
+        assertThat(result.fire(), is(true));
+        assertThat(result.isNew(), is(true));
+
+        writeState(runContext, key, state, ttl);
+        var reloaded = readState(runContext, key, ttl);
+
+        assertThat(reloaded, hasKey("gs://bucket/file1.csv"));
+        assertThat(reloaded.get("gs://bucket/file1.csv").version(), is("v1"));
+
+        var result2 = computeAndUpdateState(reloaded, candidate, StatefulTriggerInterface.On.CREATE_OR_UPDATE);
+        assertThat(result2.fire(), is(false));
+    }
+
+    @Test
+    void shouldExpireOldEntriesAfterTTL() throws Exception {
+        var flow = Flow.builder()
+            .namespace("io.kestra.unittest")
+            .id("test-flow")
+            .revision(1)
+            .build();
+
+        var runContext = runContextFactory.of(flow, Map.of(
+            "flow", Map.of(
+                "tenantId", "main",
+                "namespace", "io.kestra.unittest",
+                "id", "test-flow",
+                "revision", 1
+            )
+        ));
+
+        var key = defaultKey("ns", "test-flow", "trigger-ttl");
+        var ttl = Optional.of(Duration.ofMinutes(5));
+        var now = Instant.now();
+
+        var state = new HashMap<String, Entry>();
+        state.put("gs://bucket/old.csv", new Entry("gs://bucket/old.csv", "v1", now.minus(Duration.ofHours(2)), now.minus(Duration.ofHours(2))));
+        state.put("gs://bucket/new.csv", new Entry("gs://bucket/new.csv", "v1", now, now));
+
+        writeState(runContext, key, state, ttl);
+        var reloaded = readState(runContext, key, ttl);
+
+        assertThat(reloaded, allOf(hasKey("gs://bucket/new.csv"), not(hasKey("gs://bucket/old.csv"))));
+    }
+}

core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java

@@ -285,7 +285,7 @@ public abstract class AbstractRunnerTest {
     @LoadFlows(value = {"flows/valids/switch.yaml",
         "flows/valids/task-flow.yaml",
         "flows/valids/task-flow-inherited-labels.yaml"}, tenantId = TENANT_1)
-    void flowWaitFailed() throws Exception {
+    public void flowWaitFailed() throws Exception {
         flowCaseTest.waitFailed(TENANT_1);
     }
 

core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java

@@ -333,7 +333,7 @@ class ExecutionServiceTest {
         assertThat(restart.findTaskRunByTaskIdAndValue("1_each", List.of()).getState().getCurrent()).isEqualTo(State.Type.RUNNING);
         assertThat(restart.findTaskRunByTaskIdAndValue("2-1_seq", List.of("value 1")).getState().getCurrent()).isEqualTo(State.Type.FAILED);
         assertThat(restart.findTaskRunByTaskIdAndValue("2-1_seq", List.of("value 1")).getState().getHistories()).hasSize(4);
-        assertThat(restart.findTaskRunByTaskIdAndValue("2-1_seq", List.of("value 1")).getAttempts()).isNull();
+        assertThat(restart.findTaskRunByTaskIdAndValue("2-1_seq", List.of("value 1")).getAttempts().getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
 
         restart = executionService.markAs(execution, flow, execution.findTaskRunByTaskIdAndValue("2-1-2_t2", List.of("value 1")).getId(), State.Type.FAILED);
 

core/src/test/java/io/kestra/core/runners/FlowInputOutputTest.java

@@ -8,6 +8,7 @@ import io.kestra.core.models.flows.Type;
 import io.kestra.core.models.flows.input.FileInput;
 import io.kestra.core.models.flows.input.InputAndValue;
 import io.kestra.core.models.flows.input.IntInput;
+import io.kestra.core.models.flows.input.MultiselectInput;
 import io.kestra.core.models.flows.input.StringInput;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.secret.SecretNotFoundException;
@@ -318,6 +319,24 @@ class FlowInputOutputTest {
         Assertions.assertEquals("******", results.getFirst().value());
     }
     
+    @Test
+    void shouldNotObfuscateSecretsInSelectWhenValidatingInputs() {
+        // Given
+        MultiselectInput input = MultiselectInput.builder()
+            .id("input")
+            .type(Type.MULTISELECT)
+            .expression("{{ [secret('???')] }}")
+            .required(false)
+            .build();
+        
+        // When
+        List<InputAndValue> results = flowInputOutput.validateExecutionInputs(List.of(input), null, DEFAULT_TEST_EXECUTION, Mono.empty()).block();
+        
+        // Then
+        Assertions.assertEquals(TEST_SECRET_VALUE, ((MultiselectInput)results.getFirst().input()).getValues().getFirst());
+    }
+    
+    
     @Test
     void shouldNotObfuscateSecretsWhenReadingInputs() {
         // Given

core/src/test/java/io/kestra/core/runners/pebble/functions/KvFunctionTest.java

@@ -1,26 +1,26 @@
 package io.kestra.core.runners.pebble.functions;
 
-import static io.kestra.core.runners.pebble.functions.FunctionTestUtils.getVariables;
-import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
-import static org.assertj.core.api.Assertions.assertThat;
-
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.runners.VariableRenderer;
-import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.storages.kv.InternalKVStore;
+import io.kestra.core.storages.kv.KVMetadata;
 import io.kestra.core.storages.kv.KVStore;
 import io.kestra.core.storages.kv.KVValueAndMetadata;
 import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
-import java.io.IOException;
-import java.net.URI;
-import java.util.Map;
 import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Map;
+
+import static io.kestra.core.runners.pebble.functions.FunctionTestUtils.getVariables;
+import static org.assertj.core.api.Assertions.assertThat;
+
 @KestraTest(startRunner = true)
 public class KvFunctionTest {
 
@@ -108,6 +108,25 @@ public class KvFunctionTest {
         assertThat(rendered).isEqualTo("");
     }
 
+    @Test
+    void shouldThrowOrGetEmptyIfExpiredDependingOnErrorOnMissing() throws IOException, IllegalVariableEvaluationException {
+        String tenant = TestsUtils.randomTenant();
+        String namespace = TestsUtils.randomNamespace();
+        Map<String, Object> variables = getVariables(tenant, namespace);
+
+        KVStore kv = new InternalKVStore(tenant, namespace, storageInterface);
+        kv.put("my-expired-key", new KVValueAndMetadata(new KVMetadata(null, Instant.now().minus(1, ChronoUnit.HOURS)), "anyValue"));
+
+        String rendered = variableRenderer.render("{{ kv('my-expired-key', errorOnMissing=false) }}", variables);
+        assertThat(rendered).isEqualTo("");
+
+        kv.put("another-expired-key", new KVValueAndMetadata(new KVMetadata(null, Instant.now().minus(1, ChronoUnit.HOURS)), "anyValue"));
+
+        IllegalVariableEvaluationException exception = Assertions.assertThrows(IllegalVariableEvaluationException.class, () -> variableRenderer.render("{{ kv('another-expired-key') }}", variables));
+
+        assertThat(exception.getMessage()).isEqualTo("io.pebbletemplates.pebble.error.PebbleException: The requested value has expired ({{ kv('another-expired-key') }}:1)");
+    }
+
     @Test
     void shouldFailGivenNonExistingKeyAndErrorOnMissingTrue() {
         // Given
@@ -129,9 +148,7 @@ public class KvFunctionTest {
         String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
         Map<String, Object> variables = getVariables(tenant, "io.kestra.tests");
         // When
-        IllegalVariableEvaluationException exception = Assertions.assertThrows(IllegalVariableEvaluationException.class, () -> {
-            variableRenderer.render("{{ kv('my-key') }}", variables);
-        });
+        IllegalVariableEvaluationException exception = Assertions.assertThrows(IllegalVariableEvaluationException.class, () -> variableRenderer.render("{{ kv('my-key') }}", variables));
 
         // Then
         assertThat(exception.getMessage()).isEqualTo("io.pebbletemplates.pebble.error.PebbleException: The key 'my-key' does not exist in the namespace 'io.kestra.tests'. ({{ kv('my-key') }}:1)");

core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java

@@ -158,7 +158,7 @@ class PluginDefaultServiceTest {
               forced: false
               values:
                 expression: "{{ test }}"
-                  """,
+            """,
             DefaultTester.class.getName(),
             DefaultTriggerTester.class.getName(),
             Expression.class.getName()

core/src/test/java/io/kestra/core/test/AssertionTest.java

@@ -59,7 +59,8 @@ class AssertionTest {
     @Test
     void shouldBrokenAssert_returnError() {
         var assertion = Assertion.builder()
-            .value(new Property<>("{{ invalid-pebble-expression() }}"))
+            .value(Property.ofExpression("{{ invalid-pebble-expression() }}")
+            )
             .equalTo(Property.ofValue("value"))
             .build();
 
@@ -78,7 +79,7 @@ class AssertionTest {
     @Test
     void shouldRender_values_fromTaskOutputs() {
         var assertion = Assertion.builder()
-            .value(new Property<>("{{ outputs.my_task.res }}"))
+            .value(Property.ofExpression("{{ outputs.my_task.res }}"))
             .equalTo(Property.ofValue("value1"))
             .build();
         var runContext = runContextFactory.of(Map.of("outputs", Map.of("my_task", Map.of("res", "value1"))));
@@ -92,7 +93,7 @@ class AssertionTest {
     @Test
     void shouldRender_values_fromTaskOutputs_and_produce_defaultErrorMessage() {
         var assertion = Assertion.builder()
-            .value(new Property<>("{{ outputs.my_task.res }}"))
+            .value(Property.ofExpression("{{ outputs.my_task.res }}"))
             .equalTo(Property.ofValue("expectedValue2"))
             .build();
         var runContext = runContextFactory.of(Map.of("outputs", Map.of("my_task", Map.of("res", "actualValue1"))));
@@ -411,4 +412,4 @@ class AssertionTest {
             .first()
             .extracting(AssertionResult::isSuccess).isEqualTo(false);
     }
-}
+}

core/src/test/java/io/kestra/core/topologies/FlowTopologyServiceTest.java

@@ -160,7 +160,7 @@ class FlowTopologyServiceTest {
                                     .in(Property.ofValue(List.of(State.Type.SUCCESS)))
                                     .build(),
                                 "variables", Expression.builder()
-                                    .expression(new Property<>("{{ true }}"))
+                                    .expression(Property.ofExpression("{{ true }}"))
                                     .build()
                             ))
                             .build()

core/src/test/java/io/kestra/plugin/core/condition/PublicHolidayTest.java

@@ -1,9 +1,11 @@
 package io.kestra.plugin.core.condition;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.property.Property;
+import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.utils.TestsUtils;
 import io.kestra.core.junit.annotations.KestraTest;
@@ -11,12 +13,17 @@ import jakarta.inject.Inject;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
+import java.util.Map;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 @KestraTest
 class PublicHolidayTest {
     @Inject
     ConditionService conditionService;
+    @Inject
+    private RunContextFactory runContextFactory;
 
     @Test
     void valid() {
@@ -55,6 +62,44 @@ class PublicHolidayTest {
         assertThat(conditionService.isValid(publicHoliday, flow, execution)).isFalse();
     }
 
+    @Test
+    void validWithDynamicRender() {
+        Flow flow = TestsUtils.mockFlow();
+
+        Map<String, Object> variables = Map.of(
+            "trigger", Map.of("date", "2023-07-14")
+        );
+        Execution execution = TestsUtils.mockExecution(flow, ImmutableMap.of());
+        PublicHoliday publicHoliday = PublicHoliday.builder()
+            .country(Property.ofValue("FR"))
+            .build();
+       ConditionContext conditionContext= ConditionContext.builder()
+            .flow(flow)
+            .execution(execution)
+            .runContext(runContextFactory.of(flow,execution))
+            .variables(variables)
+            .build();
+        assertThat(conditionService.valid(flow, Collections.singletonList(publicHoliday), conditionContext)).isTrue();
+    }
+    @Test
+    void invalidWithDynamicRender() {
+        Flow flow = TestsUtils.mockFlow();
+
+        Map<String, Object> variables = Map.of(
+            "trigger", Map.of("date", "2023-01-02")
+        );
+        Execution execution = TestsUtils.mockExecution(flow, ImmutableMap.of());
+        PublicHoliday publicHoliday = PublicHoliday.builder()
+            .country(Property.ofValue("FR"))
+            .build();
+        ConditionContext conditionContext= ConditionContext.builder()
+            .flow(flow)
+            .execution(execution)
+            .runContext(runContextFactory.of(flow,execution))
+            .variables(variables)
+            .build();
+        assertThat(conditionService.valid(flow, Collections.singletonList(publicHoliday), conditionContext)).isFalse();
+    }
     @Test
     @Disabled("Locale is not deterministic on CI")
     void disabled() {

core/src/test/java/io/kestra/plugin/core/flow/IfTest.java

@@ -4,6 +4,7 @@ import io.kestra.core.junit.annotations.ExecuteFlow;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.junit.annotations.LoadFlows;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.TaskRunAttempt;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.runners.TestRunnerUtils;
@@ -11,6 +12,7 @@ import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
@@ -30,11 +32,15 @@ class IfTest {
     void ifTruthy() throws TimeoutException, QueueException {
         Execution execution = runnerUtils.runOne(TENANT_ID, "io.kestra.tests", "if-condition", null,
             (f, e) -> Map.of("param", true) , Duration.ofSeconds(120));
+        List<TaskRunAttempt> flowableAttempts=execution.findTaskRunsByTaskId("if").getFirst().getAttempts();
 
         assertThat(execution.getTaskRunList()).hasSize(2);
         assertThat(execution.findTaskRunsByTaskId("when-true").getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
 
+        assertThat(flowableAttempts).isNotNull();
+        assertThat(flowableAttempts.getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+
         execution = runnerUtils.runOne(TENANT_ID, "io.kestra.tests", "if-condition", null,
             (f, e) -> Map.of("param", "true") , Duration.ofSeconds(120));
 

core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java

@@ -56,18 +56,21 @@ public class PauseTest {
 
     @FlakyTest(description = "This test is too flaky and it always pass in JDBC and Kafka")
     @Test
+    @LoadFlows("flows/valids/pause-delay.yaml")
     void delay() throws Exception {
         suite.runDelay(runnerUtils);
     }
 
     @FlakyTest(description = "This test is too flaky and it always pass in JDBC and Kafka")
     @Test
+    @LoadFlows("flows/valids/pause-duration-from-input.yaml")
     void delayFromInput() throws Exception {
         suite.runDurationFromInput(runnerUtils);
     }
 
     @FlakyTest(description = "This test is too flaky and it always pass in JDBC and Kafka")
     @Test
+    @LoadFlows("flows/valids/each-parallel-pause.yml")
     void parallelDelay() throws Exception {
         suite.runParallelDelay(runnerUtils);
     }

core/src/test/java/io/kestra/plugin/core/flow/RuntimeLabelsTest.java

@@ -8,6 +8,8 @@ import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.junit.annotations.LoadFlows;
 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.executions.TaskRunAttempt;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.runners.TestRunnerUtils;
@@ -58,6 +60,14 @@ class RuntimeLabelsTest {
             new Label("keyFromList", "valueFromList"),
             new Label("keyFromExecution", "valueFromExecution"),
             new Label("overriddenExecutionLabelKey", labelsOverriderTaskRunId));
+
+        TaskRun labelTaskRun = execution.findTaskRunsByTaskId("override-labels").getFirst();
+        TaskRunAttempt labelRunAttempt = labelTaskRun.lastAttempt();
+
+        assertThat(labelRunAttempt.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(labelRunAttempt.getState().getHistories().size()).isEqualTo(3);
+        assertThat(labelRunAttempt.getState().getHistories()).extracting(State.History::getState)
+            .containsExactly(State.Type.CREATED, State.Type.RUNNING, State.Type.SUCCESS);
     }
 
 
@@ -69,6 +79,15 @@ class RuntimeLabelsTest {
 
         String labelsTaskRunId = execution.findTaskRunsByTaskId("labels").getFirst().getId();
         assertThat(execution.getLabels()).contains(new Label("someLabel", labelsTaskRunId));
+
+        TaskRun labelTaskRun = execution.findTaskRunsByTaskId("labels").getFirst();
+        TaskRunAttempt labelRunAttempt = labelTaskRun.lastAttempt();
+
+        assertThat(labelRunAttempt.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(labelRunAttempt.getState().getHistories().size()).isEqualTo(3);
+        assertThat(labelRunAttempt.getState().getHistories()).extracting(State.History::getState)
+            .containsExactly(State.Type.CREATED, State.Type.RUNNING, State.Type.SUCCESS);
+
     }
 
     @Test
@@ -102,6 +121,15 @@ class RuntimeLabelsTest {
             new Label("floatValue", "3.14"),
             new Label("taskRunId", labelsTaskRunId),
             new Label("existingLabel", "someValue"));
+
+        TaskRun labelTaskRun = execution.findTaskRunsByTaskId("update-labels").getFirst();
+        TaskRunAttempt labelRunAttempt = labelTaskRun.lastAttempt();
+
+        assertThat(labelRunAttempt.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(labelRunAttempt.getState().getHistories().size()).isEqualTo(3);
+        assertThat(labelRunAttempt.getState().getHistories()).extracting(State.History::getState)
+            .containsExactly(State.Type.CREATED, State.Type.RUNNING, State.Type.SUCCESS);
+
     }
 
     @Test
@@ -136,6 +164,14 @@ class RuntimeLabelsTest {
             new Label("boolValue", "true"),
             new Label("floatValue", "3.14"),
             new Label("taskRunId", labelsTaskRunId));
+
+        TaskRun labelTaskRun = execution.findTaskRunsByTaskId("update-labels").getFirst();
+        TaskRunAttempt labelRunAttempt = labelTaskRun.lastAttempt();
+
+        assertThat(labelRunAttempt.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(labelRunAttempt.getState().getHistories().size()).isEqualTo(3);
+        assertThat(labelRunAttempt.getState().getHistories()).extracting(State.History::getState)
+            .containsExactly(State.Type.CREATED, State.Type.RUNNING, State.Type.SUCCESS);
     }
 
     @Test
@@ -159,6 +195,14 @@ class RuntimeLabelsTest {
             new Label("fromStringKey", "value2"),
             new Label("fromListKey", "value2")
         );
+
+        TaskRun labelTaskRun = execution.findTaskRunsByTaskId("from-string").getFirst();
+        TaskRunAttempt labelRunAttempt = labelTaskRun.lastAttempt();
+
+        assertThat(labelRunAttempt.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
+        assertThat(labelRunAttempt.getState().getHistories().size()).isEqualTo(3);
+        assertThat(labelRunAttempt.getState().getHistories()).extracting(State.History::getState)
+            .containsExactly(State.Type.CREATED, State.Type.RUNNING, State.Type.SUCCESS);
     }
 
     @Test
@@ -180,5 +224,11 @@ class RuntimeLabelsTest {
         assertThat(execution.getLabels()).containsExactly(
             new Label(Label.CORRELATION_ID, execution.getId())
         );
+
+        TaskRun labelTaskRun = execution.findTaskRunsByTaskId("from-string").getFirst();
+        TaskRunAttempt labelRunAttempt = labelTaskRun.lastAttempt();
+
+        assertThat(labelRunAttempt.getState().getCurrent()).isEqualTo(State.Type.FAILED);
+        assertThat(labelRunAttempt.getState().getHistories().size()).isEqualTo(1);
     }
 }

core/src/test/java/io/kestra/plugin/core/flow/SequentialTest.java

@@ -5,22 +5,32 @@ import static org.assertj.core.api.Assertions.assertThat;
 import io.kestra.core.junit.annotations.ExecuteFlow;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.TaskRunAttempt;
 import io.kestra.core.models.flows.State;
 import org.junit.jupiter.api.Test;
+import java.util.List;
 
 @KestraTest(startRunner = true)
 class SequentialTest {
     @Test
     @ExecuteFlow("flows/valids/sequential.yaml")
     void sequential(Execution execution) {
+        List<TaskRunAttempt> flowableAttempts=execution.findTaskRunsByTaskId("1-seq").getFirst().getAttempts();
+
         assertThat(execution.getTaskRunList()).hasSize(11);
+        assertThat(flowableAttempts).isNotNull();
+        assertThat(flowableAttempts.getFirst().getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.SUCCESS);
     }
 
     @Test
     @ExecuteFlow("flows/valids/sequential-with-global-errors.yaml")
     void sequentialWithGlobalErrors(Execution execution) {
+        List<TaskRunAttempt> flowableAttempts=execution.findTaskRunsByTaskId("parent-seq").getFirst().getAttempts();
+
         assertThat(execution.getTaskRunList()).hasSize(6);
+        assertThat(flowableAttempts).isNotNull();
+        assertThat(flowableAttempts.getFirst().getState().getCurrent()).isEqualTo(State.Type.FAILED);
         assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.FAILED);
     }
 

core/src/test/java/io/kestra/plugin/core/kv/SetTest.java

@@ -194,8 +194,8 @@ class SetTest {
         KVStoreException exception = Assertions.assertThrows(KVStoreException.class, () -> Set.builder()
             .id(Set.class.getSimpleName())
             .type(Set.class.getName())
-            .key(new Property<>("{{ inputs.key }}"))
-            .value(new Property<>("{{ inputs.value }}"))
+            .key(Property.ofExpression("{{ inputs.key }}"))
+            .value(Property.ofExpression("{{ inputs.value }}"))
             .overwrite(Property.ofValue(false))
             .build().run(runContext));
         assertThat(exception.getMessage()).isEqualTo("Cannot set value for key '%s'. Key already exists and `overwrite` is set to `false`.".formatted(key));
@@ -239,4 +239,4 @@ class SetTest {
         set.run(runContext);
         return runContext.namespaceKv(runContext.flowInfo().namespace());
     }
-}
+}

core/src/test/java/io/kestra/plugin/core/metric/PublishTest.java

@@ -4,6 +4,7 @@ import io.kestra.core.context.TestRunContextFactory;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.metrics.CounterMetric;
+import io.kestra.core.models.tasks.metrics.GaugeMetric;
 import io.kestra.core.models.tasks.metrics.TimerMetric;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.utils.TestsUtils;
@@ -35,6 +36,10 @@ public class PublishTest {
                     TimerMetric.builder()
                         .value(Property.ofValue(Duration.parse("PT5H")))
                         .name(Property.ofValue("timer"))
+                        .build(),
+                    GaugeMetric.builder()
+                        .value(Property.ofValue(1.0))
+                        .name(Property.ofValue("gauge"))
                         .build()
                 ))
             )
@@ -44,7 +49,7 @@ public class PublishTest {
 
         publish.run(runContext);
 
-        assertThat(runContext.metrics().size()).isEqualTo(2);
+        assertThat(runContext.metrics().size()).isEqualTo(3);
     }
 
 }

core/src/test/java/io/kestra/plugin/core/trigger/PollingTest.java

@@ -22,6 +22,7 @@ class PollingTest {
         assertThat(optionalExecution).isPresent();
         Execution execution = optionalExecution.get();
         assertThat(execution.getFlowId()).isEqualTo("polling-flow");
+        assertThat(execution.getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
         assertTrue(execution.getState().getCurrent().isCreated());
     }
 }

core/src/test/java/io/kestra/plugin/core/trigger/ScheduleTest.java

@@ -66,6 +66,7 @@ class ScheduleTest {
             .id(IdUtils.create())
             .namespace("io.kestra.unittest")
             .revision(1)
+            .variables(Map.of("custom_var", "VARIABLE VALUE"))
             .tasks(Collections.singletonList(Return.builder()
                 .id("test")
                 .type(Return.class.getName())
@@ -101,7 +102,7 @@ class ScheduleTest {
         assertThat(evaluate.isPresent()).isTrue();
         assertThat(evaluate.get().getLabels()).hasSize(3);
         assertTrue(evaluate.get().getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
-
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
         var vars = evaluate.get().getTrigger().getVariables();
         var inputs = evaluate.get().getInputs();
 
@@ -135,7 +136,7 @@ class ScheduleTest {
         assertThat(evaluate.isPresent()).isTrue();
         assertThat(evaluate.get().getLabels()).hasSize(3);
         assertTrue(evaluate.get().getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
-
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
         var inputs = evaluate.get().getInputs();
 
         assertThat(inputs.size()).isEqualTo(2);
@@ -167,6 +168,7 @@ class ScheduleTest {
         Optional<Execution> evaluate = scheduleTrigger.evaluate(conditionContext, triggerContext);
 
         assertThat(evaluate.isPresent()).isTrue();
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
         assertThat(evaluate.get().getLabels()).contains(new Label("trigger-label-1", "trigger-label-1"));
         assertThat(evaluate.get().getLabels()).contains(new Label("trigger-label-2", "trigger-label-2"));
         assertThat(evaluate.get().getLabels()).doesNotContain(new Label("trigger-label-3", ""));
@@ -188,8 +190,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
 
         assertThat(dateFromVars((String) vars.get("date"), date)).isEqualTo(date);
         assertThat(dateFromVars((String) vars.get("next"), date)).isEqualTo(date.plus(Duration.ofMinutes(1)));
@@ -210,8 +212,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
 
         assertThat(dateFromVars((String) vars.get("date"), date)).isEqualTo(date);
         assertThat(dateFromVars((String) vars.get("next"), date)).isEqualTo(date.plus(Duration.ofSeconds(1)));
@@ -232,7 +234,6 @@ class ScheduleTest {
             ).build();
         // When
         Optional<Execution> result = trigger.evaluate(conditionContext(trigger), triggerContext);
-
         // Then
         assertThat(result.isEmpty()).isTrue();
     }
@@ -296,8 +297,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
         assertThat(dateFromVars((String) vars.get("date"), expexted)).isEqualTo(expexted);
         assertThat(dateFromVars((String) vars.get("next"), expexted)).isEqualTo(expexted.plusMonths(1));
         assertThat(dateFromVars((String) vars.get("previous"), expexted)).isEqualTo(expexted.minusMonths(1));
@@ -330,8 +331,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
         assertThat(dateFromVars((String) vars.get("date"), date)).isEqualTo(date);
         assertThat(dateFromVars((String) vars.get("next"), next)).isEqualTo(next);
         assertThat(dateFromVars((String) vars.get("previous"), previous)).isEqualTo(previous);
@@ -362,8 +363,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
         assertThat(dateFromVars((String) vars.get("date"), date)).isEqualTo(date);
         assertThat(dateFromVars((String) vars.get("previous"), previous)).isEqualTo(previous);
         assertThat(vars.containsKey("next")).isFalse();
@@ -412,7 +413,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
         assertThat(dateFromVars((String) vars.get("date"), date)).isEqualTo(date);
     }
 
@@ -437,8 +439,8 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-
-        var vars = evaluate.get().getTrigger().getVariables();;
+        assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
+        var vars = evaluate.get().getTrigger().getVariables();
 
         assertThat(dateFromVars((String) vars.get("date"), date)).isEqualTo(date);
         assertThat(ZonedDateTime.parse((String) vars.get("date")).getZone().getId()).isEqualTo("-04:00");
@@ -467,6 +469,7 @@ class ScheduleTest {
 
         // Then
         assertThat(result.isPresent()).isTrue();
+        assertThat(result.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
     }
 
     private ConditionContext conditionContext(AbstractTrigger trigger) {
@@ -479,6 +482,7 @@ class ScheduleTest {
                     new Label("flow-label-2", "flow-label-2")
                 )
             )
+            .variables(Map.of("custom_var", "VARIABLE VALUE"))
             .inputs(List.of(
                 StringInput.builder().id("input1").type(Type.STRING).required(false).build(),
                 StringInput.builder().id("input2").type(Type.STRING).defaults(Property.ofValue("default")).build()

core/src/test/resources/flows/tests/trigger-polling.yaml

@@ -1,5 +1,9 @@
 id: polling-flow
 namespace: io.kestra.tests
+
+variables:
+  custom_var: VARIABLE VALUE
+
 triggers:
   - id: polling-trigger-1
     type: io.kestra.core.tasks.test.PollingTrigger

core/src/test/resources/flows/valids/webhook-inputs.yaml

@@ -0,0 +1,15 @@
+id: webhook-inputs
+namespace: io.kestra.tests
+inputs:
+  - id: body
+    type: STRING
+tasks:
+  - id: out
+    type: io.kestra.plugin.core.debug.Return
+    format: "{{ inputs.body }}"
+triggers:
+  - id: webhook
+    type: io.kestra.plugin.core.trigger.Webhook
+    key: webhookKey
+    inputs:
+      body: "{{ trigger.body }}"

docker-compose-dind.yml

@@ -1,3 +1,8 @@
+# For ubuntu 24.04+ users, to make dind work properly with apparmor, you might need to disable the restriction on unprivileged user namespaces.
+# if `sudo sysctl kernel.apparmor_restrict_unprivileged_userns` returns `1`, you need to disable it to start dind:
+# echo 'kernel.apparmor_restrict_unprivileged_userns=0' | sudo tee /etc/sysctl.d/99-apparmor-userns.conf
+# sudo sysctl --system
+
 volumes:
   postgres-data:
     driver: local
@@ -28,12 +33,12 @@ services:
     privileged: true
     user: "1000"
     environment:
-      DOCKER_HOST: unix://dind/docker.sock
+      DOCKER_HOST: unix:///home/rootless/docker.sock
     command:
       - --log-level=fatal
       - --group=1000
     volumes:
-      - dind-socket:/dind
+      - dind-socket:/home/rootless/
       - tmp-data:/tmp/kestra-wd
 
   kestra:
@@ -55,10 +60,6 @@ services:
             username: kestra
             password: k3str4
         kestra:
-          server:
-            basic-auth:
-              username: "admin@kestra.io" # it must be a valid email address
-              password: kestra
           repository:
             type: postgres
           storage:

executor/src/main/java/io/kestra/executor/ExecutorService.java

@@ -247,7 +247,7 @@ public class ExecutorService {
             // first find the normal ended child tasks and send result
             Optional<State.Type> state;
             try {
-                state = flowableParent.resolveState(runContext, execution, parentTaskRun);
+                 state = flowableParent.resolveState(runContext, execution, parentTaskRun);
             } catch (Exception e) {
                 // This will lead to the next task being still executed, but at least Kestra will not crash.
                 // This is the best we can do, Flowable task should not fail, so it's a kind of panic mode.
@@ -268,9 +268,17 @@ public class ExecutorService {
                         Output outputs = flowableParent.outputs(runContext);
                         Map<String, Object> outputMap = MapUtils.merge(workerTaskResult.getTaskRun().getOutputs(), outputs == null ? null : outputs.toMap());
                         Variables variables = variablesService.of(StorageContext.forTask(workerTaskResult.getTaskRun()), outputMap);
+                        /// flowable attempt state transition to terminated
+                            List<TaskRunAttempt> attempts = Optional.ofNullable(parentTaskRun.getAttempts())
+                                .map(ArrayList::new)
+                                .orElseGet(ArrayList::new);
+                            State.Type endedState=endedTask.get().getTaskRun().getState().getCurrent();
+                            TaskRunAttempt updated = attempts.getLast().withState(endedState);
+                            attempts.set( attempts.size() - 1, updated);
                         return Optional.of(new WorkerTaskResult(workerTaskResult
                             .getTaskRun()
                             .withOutputs(variables)
+                            .withAttempts(attempts)
                         ));
                     } catch (Exception e) {
                         runContext.logger().error("Unable to resolve outputs from the Flowable task: {}", e.getMessage(), e);
@@ -320,7 +328,6 @@ public class ExecutorService {
 
     private List<TaskRun> childNextsTaskRun(Executor executor, TaskRun parentTaskRun) throws InternalException {
         Task parent = executor.getFlow().findTaskByTaskId(parentTaskRun.getTaskId());
-
         if (parent instanceof FlowableTask<?> flowableParent) {
             // Count the number of flowable tasks executions, some flowable are being called multiple times,
             // so this is not exactly the number of flowable taskruns but the number of times they are executed.
@@ -375,6 +382,7 @@ public class ExecutorService {
                     Output outputs = flowableTask.outputs(runContext);
                     Variables variables = variablesService.of(StorageContext.forTask(taskRun), outputs);
                     taskRun = taskRun.withOutputs(variables);
+
                 } catch (Exception e) {
                     runContext.logger().warn("Unable to save output on taskRun '{}'", taskRun, e);
                 }
@@ -1065,14 +1073,12 @@ public class ExecutorService {
 
         executor.getWorkerTasks()
             .removeIf(workerTask -> {
-                if (!(workerTask.getTask() instanceof ExecutionUpdatableTask)) {
+                if (!(workerTask.getTask() instanceof ExecutionUpdatableTask executionUpdatingTask)) {
                     return false;
                 }
 
-                var executionUpdatingTask = (ExecutionUpdatableTask) workerTask.getTask();
-
                 try {
-                    // handle runIf
+                    // Skip task if runIf condition is false
                     if (!TruthUtils.isTruthy(workerTask.getRunContext().render(workerTask.getTask().getRunIf()))) {
                         executor.withExecution(
                             executor
@@ -1083,19 +1089,28 @@ public class ExecutorService {
                         return false;
                     }
 
+                    TaskRun runningTaskRun = workerTask
+                        .getTaskRun()
+                        .withAttempts(List.of(TaskRunAttempt.builder().state(new State().withState(State.Type.RUNNING)).build()))
+                        .withState(State.Type.RUNNING);
+
                     executor.withExecution(
                         executionUpdatingTask.update(executor.getExecution(), workerTask.getRunContext())
-                            .withTaskRun(workerTask.getTaskRun().withState(State.Type.RUNNING)),
+                            .withTaskRun(runningTaskRun),
                         "handleExecutionUpdatingTask.updateExecution"
                     );
 
-                    var taskState = executionUpdatingTask.resolveState(workerTask.getRunContext(), executor.getExecution()).orElse(State.Type.SUCCESS);
+                    var terminalState = executionUpdatingTask
+                        .resolveState(workerTask.getRunContext(), executor.getExecution())
+                        .orElse(State.Type.SUCCESS);
+
+                    TaskRunAttempt terminalAttempt = runningTaskRun.lastAttempt().withState(terminalState);
+
                     workerTaskResults.add(
                         WorkerTaskResult.builder()
-                            .taskRun(workerTask.getTaskRun().withAttempts(
-                                        Collections.singletonList(TaskRunAttempt.builder().state(new State().withState(taskState)).build())
-                                    )
-                                    .withState(taskState)
+                            .taskRun(runningTaskRun
+                                .withAttempts(List.of(terminalAttempt))
+                                .withState(terminalState)
                             )
                             .build()
                     );

jdbc-mysql/src/main/resources/migrations/mysql/V1_46__taskrun_submitted.sql

@@ -13,5 +13,6 @@ ALTER TABLE executions MODIFY COLUMN `state_current` ENUM (
     'RETRYING',
     'RETRIED',
     'SKIPPED',
+    'BREAKPOINT',
     'SUBMITTED'
-    ) GENERATED ALWAYS AS (value ->> '$.state.current') STORED NOT NULL;
+    ) GENERATED ALWAYS AS (value ->> '$.state.current') STORED NOT NULL;

jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcRepository.java

@@ -58,7 +58,11 @@ public abstract class AbstractJdbcRepository {
 
     protected Condition defaultFilter(String tenantId, boolean allowDeleted) {
         var tenant = buildTenantCondition(tenantId);
-        return allowDeleted ? tenant : tenant.and(field("deleted", Boolean.class).eq(false));
+
+        // Always include `deleted` in the query filters as most database optimizers can only use and index if the leftmost columns are used in the query
+        return allowDeleted ?
+            tenant.and(field("deleted", Boolean.class).in(true, false)) :
+            tenant.and(field("deleted", Boolean.class).eq(false));
     }
 
     protected Condition defaultFilterWithNoACL(String tenantId) {
@@ -67,7 +71,11 @@ public abstract class AbstractJdbcRepository {
 
     protected Condition defaultFilterWithNoACL(String tenantId, boolean deleted) {
         var tenant = buildTenantCondition(tenantId);
-        return deleted ? tenant : tenant.and(field("deleted", Boolean.class).eq(false));
+
+        // Always include `deleted` in the query filters as most database optimizers can only use and index if the leftmost columns are used in the query
+        return deleted ?
+            tenant.and(field("deleted", Boolean.class).in(true, false)) :
+            tenant.and(field("deleted", Boolean.class).eq(false));
     }
 
     protected Condition buildTenantCondition(String tenantId) {

jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java

@@ -658,8 +658,24 @@ public class JdbcExecutor implements ExecutorInterface {
                                                     workerTaskResults.add(new WorkerTaskResult(taskRun));
                                                 }
                                             }
+                                            /// flowable attempt state transition to running
                                             if (workerTask.getTask().isFlowable()) {
-                                                workerTaskResults.add(new WorkerTaskResult(workerTask.getTaskRun().withState(State.Type.RUNNING)));
+                                                List<TaskRunAttempt> attempts = Optional.ofNullable(workerTask.getTaskRun().getAttempts())
+                                                    .map(ArrayList::new)
+                                                    .orElseGet(ArrayList::new);
+
+
+                                                attempts.add(
+                                                    TaskRunAttempt.builder()
+                                                        .state(new State().withState(State.Type.RUNNING))
+                                                        .build()
+                                                );
+
+                                                TaskRun updatedTaskRun = workerTask.getTaskRun()
+                                                    .withAttempts(attempts)
+                                                    .withState(State.Type.RUNNING);
+
+                                                workerTaskResults.add(new WorkerTaskResult(updatedTaskRun));
                                             }
                                         }
                                     } catch (Exception e) {

platform/build.gradle

@@ -12,15 +12,15 @@ javaPlatform {
 dependencies {
     // versions for libraries with multiple module but no BOM
     def slf4jVersion = "2.0.17"
-    def protobufVersion = "3.25.5" // Orc still uses 3.25.5 see https://github.com/apache/orc/blob/main/java/pom.xml
+    def protobufVersion = "3.25.8" // Orc still uses 3 see https://github.com/apache/orc/blob/main/java/pom.xml
     def bouncycastleVersion = "1.82"
     def mavenResolverVersion = "2.0.10"
     def jollydayVersion = "1.5.6"
     def jsonschemaVersion = "4.38.0"
     def kafkaVersion = "4.1.0"
     def opensearchVersion = "3.2.0"
-    def opensearchRestVersion = "3.2.0"
-    def flyingSaucerVersion = "10.0.0"
+    def opensearchRestVersion = "3.3.1"
+    def flyingSaucerVersion = "10.0.3"
     def jacksonVersion = "2.20.0"
     def jacksonAnnotationsVersion = "2.20"
     def jugVersion = "5.1.1"
@@ -33,9 +33,9 @@ dependencies {
     api platform("io.micronaut.platform:micronaut-platform:4.9.4")
     api platform("io.qameta.allure:allure-bom:2.30.0")
     // we define cloud bom here for GCP, Azure and AWS so they are aligned for all plugins that use them (secret, storage, oss and ee plugins)
-    api platform('com.google.cloud:libraries-bom:26.69.0')
+    api platform('com.google.cloud:libraries-bom:26.70.0')
     api platform("com.azure:azure-sdk-bom:1.3.0")
-    api platform('software.amazon.awssdk:bom:2.35.3')
+    api platform('software.amazon.awssdk:bom:2.35.11')
     api platform("dev.langchain4j:langchain4j-bom:$langchain4jVersion")
     api platform("dev.langchain4j:langchain4j-community-bom:$langchain4jCommunityVersion")
 
@@ -75,12 +75,7 @@ dependencies {
         api "org.apache.kafka:kafka-clients:$kafkaVersion"
         api "org.apache.kafka:kafka-streams:$kafkaVersion"
         // AWS CRT is not included in the AWS BOM but needed for the S3 Transfer manager
-        api 'software.amazon.awssdk.crt:aws-crt:0.39.0'
-
-        // we need at least 0.14, it could be removed when Micronaut contains a recent only version in their BOM
-        api "io.micrometer:micrometer-core:1.15.4"
-        // We need at least 6.17, it could be removed when Micronaut contains a recent only version in their BOM
-        api "io.micronaut.openapi:micronaut-openapi-bom:6.18.1"
+        api 'software.amazon.awssdk.crt:aws-crt:0.39.3'
 
         // Other libs
         api("org.projectlombok:lombok:1.18.42")
@@ -101,7 +96,7 @@ dependencies {
         api group: 'org.apache.maven.resolver', name: 'maven-resolver-connector-basic', version: mavenResolverVersion
         api group: 'org.apache.maven.resolver', name: 'maven-resolver-transport-file', version: mavenResolverVersion
         api group: 'org.apache.maven.resolver', name: 'maven-resolver-transport-apache', version: mavenResolverVersion
-        api 'com.github.oshi:oshi-core:6.9.0'
+        api 'com.github.oshi:oshi-core:6.9.1'
         api 'io.pebbletemplates:pebble:3.2.4'
         api group: 'co.elastic.logging', name: 'logback-ecs-encoder', version: '1.7.0'
         api group: 'de.focus-shift', name: 'jollyday-core', version: jollydayVersion
@@ -128,7 +123,7 @@ dependencies {
         api group: 'jakarta.annotation', name: 'jakarta.annotation-api', version: '3.0.0'
         api group: 'org.eclipse.angus', name: 'jakarta.mail', version: '2.0.5'
         api group: 'com.github.ben-manes.caffeine', name: 'caffeine', version: '3.2.2'
-        api group: 'de.siegmar', name: 'fastcsv', version: '4.0.0'
+        api group: 'de.siegmar', name: 'fastcsv', version: '4.1.0'
         // Json Diff
         api group: 'com.github.java-json-tools', name: 'json-patch', version: '1.13'
 
@@ -145,7 +140,7 @@ dependencies {
         api group: 'org.exparity', name: 'hamcrest-date', version: '2.0.8'
         api "org.wiremock:wiremock-jetty12:3.13.1"
         api "org.apache.kafka:kafka-streams-test-utils:$kafkaVersion"
-        api "com.microsoft.playwright:playwright:1.55.0"
+        api "com.microsoft.playwright:playwright:1.56.0"
         api "org.awaitility:awaitility:4.3.0"
 
         // Kestra components

scheduler/src/main/java/io/kestra/scheduler/AbstractScheduler.java

@@ -7,6 +7,7 @@ import io.kestra.core.events.CrudEvent;
 import io.kestra.core.events.CrudEventType;
 import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.exceptions.InvalidTriggerConfigurationException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.HasUID;
 import io.kestra.core.models.conditions.Condition;
@@ -283,10 +284,22 @@ public abstract class AbstractScheduler implements Scheduler {
                         workerTriggerResult.getExecution().get(),
                         workerTriggerResult.getTriggerContext()
                     );
-                    ZonedDateTime nextExecutionDate = this.nextEvaluationDate(workerTriggerResult.getTrigger());
+                    ZonedDateTime nextExecutionDate;
+                    try {
+                        nextExecutionDate = this.nextEvaluationDate(workerTriggerResult.getTrigger());
+                    } catch (InvalidTriggerConfigurationException e) {
+                        disableInvalidTrigger(workerTriggerResult.getTriggerContext(), e);
+                        return;
+                    }
                     this.handleEvaluateWorkerTriggerResult(triggerExecution, nextExecutionDate);
                 } else {
-                    ZonedDateTime nextExecutionDate = this.nextEvaluationDate(workerTriggerResult.getTrigger());
+                    ZonedDateTime nextExecutionDate;
+                    try {
+                        nextExecutionDate = this.nextEvaluationDate(workerTriggerResult.getTrigger());
+                    } catch (InvalidTriggerConfigurationException e) {
+                        disableInvalidTrigger(workerTriggerResult.getTriggerContext(), e);
+                        return;
+                    }
                     this.triggerState.update(Trigger.of(workerTriggerResult.getTriggerContext(), nextExecutionDate));
                 }
             }
@@ -450,7 +463,7 @@ public abstract class AbstractScheduler implements Scheduler {
         // by default: do nothing
     }
 
-    private ZonedDateTime nextEvaluationDate(AbstractTrigger abstractTrigger) {
+    private ZonedDateTime nextEvaluationDate(AbstractTrigger abstractTrigger) throws InvalidTriggerConfigurationException {
         if (abstractTrigger instanceof PollingTriggerInterface interval) {
             return interval.nextEvaluationDate();
         } else {
@@ -458,7 +471,7 @@ public abstract class AbstractScheduler implements Scheduler {
         }
     }
 
-    private ZonedDateTime nextEvaluationDate(AbstractTrigger abstractTrigger, ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws Exception {
+    private ZonedDateTime nextEvaluationDate(AbstractTrigger abstractTrigger, ConditionContext conditionContext, Optional<? extends TriggerContext> last) throws Exception, InvalidTriggerConfigurationException {
         if (abstractTrigger instanceof PollingTriggerInterface interval) {
             return interval.nextEvaluationDate(conditionContext, last);
         } else {
@@ -514,6 +527,10 @@ public abstract class AbstractScheduler implements Scheduler {
                             triggerContext = lastTrigger.toBuilder()
                                 .nextExecutionDate(this.nextEvaluationDate(abstractTrigger, conditionContext, Optional.of(lastTrigger)))
                                 .build();
+                        } catch (InvalidTriggerConfigurationException e) {
+                            logError(conditionContext, flow, abstractTrigger, e);
+                            disableInvalidTrigger(flow, abstractTrigger, e);
+                            return null;
                         } catch (Exception e) {
                             logError(conditionContext, flow, abstractTrigger, e);
                             return null;
@@ -537,6 +554,47 @@ public abstract class AbstractScheduler implements Scheduler {
             .filter(Objects::nonNull).toList();
     }
 
+    private void disableInvalidTrigger(TriggerContext triggerContext, Throwable e) {
+        try {
+            var disabledTrigger = Trigger.builder()
+                .tenantId(triggerContext.getTenantId())
+                .namespace(triggerContext.getNamespace())
+                .flowId(triggerContext.getFlowId())
+                .triggerId(triggerContext.getTriggerId())
+                .date(triggerContext.getDate())
+                .backfill(triggerContext.getBackfill())
+                .stopAfter(triggerContext.getStopAfter())
+                .disabled(true)
+                .updatedDate(Instant.now())
+                .build();
+
+            triggerState.update(disabledTrigger);
+
+            triggerQueue.emit(disabledTrigger);
+
+            log.warn("Disabled trigger {}.{} due to invalid configuration: {}", disabledTrigger.getFlowId(), disabledTrigger.getTriggerId(), e.getMessage());
+        } catch (Exception ex) {
+            log.error("Failed to disable trigger {}.{}: {}", triggerContext.getFlowId(), triggerContext.getTriggerId(), ex.getMessage(), ex);
+        }
+    }
+
+    private void disableInvalidTrigger(FlowWithSource flow, AbstractTrigger trigger, Throwable e) {
+        var disabledTrigger = Trigger.builder()
+            .tenantId(flow.getTenantId())
+            .namespace(flow.getNamespace())
+            .flowId(flow.getId())
+            .triggerId(trigger.getId())
+            .disabled(true)
+            .updatedDate(Instant.now())
+            .build();
+
+        disableInvalidTrigger(disabledTrigger, e);
+    }
+
+    private void disableInvalidTrigger(FlowWithWorkerTrigger f, Throwable e) {
+        disableInvalidTrigger(f.getTriggerContext(), e);
+    }
+
     abstract public void handleNext(List<FlowWithSource> flows, ZonedDateTime now, BiConsumer<List<Trigger>, ScheduleContextInterface> consumer);
 
     public List<FlowWithTriggers> schedulerTriggers() {
@@ -681,6 +739,10 @@ public abstract class AbstractScheduler implements Scheduler {
                             ZonedDateTime nextExecutionDate = null;
                             try {
                                 nextExecutionDate = this.nextEvaluationDate(f.getAbstractTrigger(), f.getConditionContext(), Optional.of(f.getTriggerContext()));
+                            } catch (InvalidTriggerConfigurationException e) {
+                                logError(f, e);
+                                disableInvalidTrigger(f, e);
+                                return;
                             } catch (Exception e) {
                                 logError(f, e);
                             }
@@ -700,7 +762,15 @@ public abstract class AbstractScheduler implements Scheduler {
                             .labels(LabelService.labelsExcludingSystem(f.getFlow()))
                             .state(new State().withState(State.Type.FAILED))
                             .build();
-                        ZonedDateTime nextExecutionDate = this.nextEvaluationDate(f.getAbstractTrigger());
+                        ZonedDateTime nextExecutionDate;
+                        try {
+                            nextExecutionDate = this.nextEvaluationDate(f.getAbstractTrigger());
+                        } catch (InvalidTriggerConfigurationException e2) {
+                            logError(f, e2);
+                            disableInvalidTrigger(f, e2);
+                            return;
+                        }
+
                         var trigger = f.getTriggerContext().resetExecution(State.Type.FAILED, nextExecutionDate);
                         this.saveLastTriggerAndEmitExecution(execution, trigger, triggerToSave -> this.triggerState.save(triggerToSave, scheduleContext, "/kestra/services/scheduler/handle/save/on-error"));
                     }

scheduler/src/test/java/io/kestra/scheduler/SchedulerPollingTriggerTest.java

@@ -1,10 +1,12 @@
 package io.kestra.scheduler;
 
 import io.kestra.core.models.Label;
-import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.conditions.ConditionContext;
+import io.kestra.core.models.flows.*;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.SchedulerTriggerStateInterface;
 import io.kestra.core.tasks.test.FailingPollingTrigger;
@@ -12,8 +14,6 @@ import io.kestra.core.utils.TestsUtils;
 import io.kestra.jdbc.runner.JdbcScheduler;
 import io.kestra.plugin.core.condition.Expression;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.State;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.runners.FlowListeners;
@@ -25,12 +25,15 @@ import io.kestra.core.utils.Await;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
 
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -150,7 +153,7 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
                 List.of(
                     Expression.builder()
                         .type(Expression.class.getName())
-                        .expression(new Property<>("{{ trigger.date | date() < now() }}"))
+                        .expression(Property.ofExpression("{{ trigger.date | date() < now() }}"))
                         .build()
                 ))
             .build();
@@ -228,6 +231,31 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
         }
     }
 
+    @Test
+    void shouldDisableTriggerOnInvalidOverflowInterval() throws Exception {
+        FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
+
+        OverflowIntervalTrigger overflow = OverflowIntervalTrigger.builder()
+            .id("overflow-interval")
+            .type(OverflowIntervalTrigger.class.getName())
+            .build();
+
+        FlowWithSource flow = createPollingTriggerFlow(overflow);
+        doReturn(List.of(flow)).when(flowListenersServiceSpy).flows();
+
+        try (
+            AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+            Worker worker = applicationContext.createBean(TestMethodScopedWorker.class, IdUtils.create(), 8, null)
+        ) {
+            worker.run();
+            scheduler.run();
+
+            Trigger key = Trigger.of(flow, overflow);
+
+            Await.until(() -> this.triggerState.findLast(key).map(TriggerContext::getDisabled).get().booleanValue(), Duration.ofMillis(100), Duration.ofSeconds(15));
+        }
+    }
+
     private FlowWithSource createPollingTriggerFlow(AbstractTrigger pollingTrigger) {
         return createFlow(Collections.singletonList(pollingTrigger));
     }
@@ -246,4 +274,20 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
             flowListenersServiceSpy
         );
     }
-}
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    public static class OverflowIntervalTrigger extends AbstractTrigger implements PollingTriggerInterface {
+        // we set a large interval which will throw an exception
+        @Builder.Default
+        private final Duration interval = Duration.ofSeconds(Long.MAX_VALUE);
+
+        @Override
+        public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) {
+            return Optional.empty();
+        }
+    }
+}

scheduler/src/test/java/io/kestra/scheduler/SchedulerScheduleTest.java

@@ -499,7 +499,7 @@ public class SchedulerScheduleTest extends AbstractSchedulerTest {
                 List.of(
                     Expression.builder()
                         .type(Expression.class.getName())
-                        .expression(new Property<>("{{ trigger.date | date() < now() }}"))
+                        .expression(Property.ofExpression("{{ trigger.date | date() < now() }}"))
                         .build()
                 )
             )

scheduler/src/test/java/io/kestra/scheduler/SchedulerTriggerChangeTest.java

@@ -80,7 +80,7 @@ public class SchedulerTriggerChangeTest extends AbstractSchedulerTest {
             .tasks(Collections.singletonList(Return.builder()
                 .id("test")
                 .type(Return.class.getName())
-                .format(new Property<>("{{ inputs.testInputs }}"))
+                .format(Property.ofExpression("{{ inputs.testInputs }}"))
                 .build())
             )
             .build();
@@ -181,4 +181,4 @@ public class SchedulerTriggerChangeTest extends AbstractSchedulerTest {
             return Optional.of(TriggerService.generateExecution(this, conditionContext, context, Map.of("sleep", sleep.toString())));
         }
     }
-}
+}