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/ISSUE_TEMPLATE/bug.yml

@@ -4,7 +4,7 @@ body:
   - type: markdown
     attributes:
       value: |
-        Thanks for reporting an issue! Please provide a [Minimal Reproducible Example](https://stackoverflow.com/help/minimal-reproducible-example) and share any additional information that may help reproduce, troubleshoot, and hopefully fix the issue, including screenshots, error traceback, and your Kestra server logs. For quick questions, you can contact us directly on [Slack](https://kestra.io/slack).
+        Thanks for reporting an issue! Please provide a [Minimal Reproducible Example](https://stackoverflow.com/help/minimal-reproducible-example) and share any additional information that may help reproduce, troubleshoot, and hopefully fix the issue, including screenshots, error traceback, and your Kestra server logs. For quick questions, you can contact us directly on [Slack](https://kestra.io/slack). Don't forget to give us a star! ⭐
   - type: textarea
     attributes:
       label: Describe the issue

.github/ISSUE_TEMPLATE/feature.yml

@@ -4,7 +4,7 @@ body:
   - type: textarea
     attributes:
       label: Feature description
-      placeholder: Tell us more about your feature request
+      placeholder: Tell us more about your feature request. Don't forget to give us a star! ⭐
     validations:
       required: true
 labels:

.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/pull_request_template.md

@@ -35,4 +35,4 @@ Remove this section if this change applies to all flows or to the documentation
 
 If there are no setup requirements, you can remove this section.
 
-Thank you for your contribution. ❤️  -->
+Thank you for your contribution. ❤️ Don't forget to give us a star! ⭐  -->

.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/codeql-analysis.yml

@@ -40,7 +40,7 @@ jobs:
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v3
+      uses: github/codeql-action/init@v4
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -58,7 +58,7 @@ jobs:
 
     - name: Setup gradle
       if: ${{ matrix.language == 'java' }}
-      uses: gradle/actions/setup-gradle@v4
+      uses: gradle/actions/setup-gradle@v5
 
     - name: Build with Gradle
       if: ${{ matrix.language == 'java' }}
@@ -68,7 +68,7 @@ jobs:
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild
       if: ${{ matrix.language != 'java' }}
-      uses: github/codeql-action/autobuild@v3
+      uses: github/codeql-action/autobuild@v4
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -82,4 +82,4 @@ jobs:
     #   make release
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v3
+      uses: github/codeql-action/analyze@v4

.github/workflows/main-build.yml

@@ -67,20 +67,24 @@ jobs:
 
   end:
     runs-on: ubuntu-latest
-    needs: [publish-develop-docker, publish-develop-maven]
-    if: always()
+    needs: [backend-tests, frontend-tests, publish-develop-docker, publish-develop-maven]
+    if: "always() && github.repository == 'kestra-io/kestra'"
     steps:
+      - run: echo "end CI of failed or success"
       - name: Trigger EE Workflow
-        uses: peter-evans/repository-dispatch@v3
-        if: github.ref == 'refs/heads/develop' && needs.release.result == '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() && env.SLACK_WEBHOOK_URL != 0 && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/main' || 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

@@ -22,12 +22,11 @@ jobs:
           fetch-depth: 0
 
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true
           node-enabled: true
-          caches-enabled: true
 
       # Npm
       - name: Npm - Install
@@ -69,11 +68,10 @@ jobs:
         with:
           java-enabled: false
           node-enabled: false
-          caches-enabled: true
 
       # 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'
@@ -83,7 +81,7 @@ jobs:
           skip-dirs: /app/plugins
 
       - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@v3
+        uses: github/codeql-action/upload-sarif@v4
         with:
           sarif_file: 'trivy-results.sarif'
           category: docker-
@@ -110,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
@@ -120,6 +118,7 @@ jobs:
           output: 'trivy-results.sarif'
 
       - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@v3
+        uses: github/codeql-action/upload-sarif@v4
         with:
-          sarif_file: 'trivy-results.sarif'
+          sarif_file: 'trivy-results.sarif'
+          category: docker-

.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

Dockerfile.pr

@@ -1,4 +1,5 @@
-FROM  kestra/kestra:develop
+ARG KESTRA_DOCKER_BASE_VERSION=develop
+FROM  kestra/kestra:$KESTRA_DOCKER_BASE_VERSION
 
 USER root
 

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.5" apply false
+    id "org.owasp.dependencycheck" version "12.1.8" apply false
 }
 
 idea {
@@ -206,41 +206,69 @@ subprojects {subProj ->
             testImplementation 'org.assertj:assertj-core'
         }
 
-        test {
-            useJUnitPlatform()
-            reports {
-                junitXml.required = true
-                junitXml.outputPerTestCase = true
-                junitXml.mergeReruns = true
-                junitXml.includeSystemErrLog = true;
-                junitXml.outputLocation = layout.buildDirectory.dir("test-results/test")
-            }
-
+        def commonTestConfig = { Test t ->
             // set Xmx for test workers
-            maxHeapSize = '4g'
+            t.maxHeapSize = '4g'
 
             // configure en_US default locale for tests
-            systemProperty 'user.language', 'en'
-            systemProperty 'user.country', 'US'
+            t.systemProperty 'user.language', 'en'
+            t.systemProperty 'user.country', 'US'
 
-            environment 'SECRET_MY_SECRET', "{\"secretKey\":\"secretValue\"}".bytes.encodeBase64().toString()
-            environment 'SECRET_NEW_LINE', "cGFzc3dvcmR2ZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2ZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2\nZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2ZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2ZXJ5dmVyeXZl\neXJsb25n"
-            environment 'SECRET_WEBHOOK_KEY', "secretKey".bytes.encodeBase64().toString()
-            environment 'SECRET_NON_B64_SECRET', "some secret value"
-            environment 'SECRET_PASSWORD', "cGFzc3dvcmQ="
-            environment 'ENV_TEST1', "true"
-            environment 'ENV_TEST2', "Pass by env"
+            t.environment 'SECRET_MY_SECRET', "{\"secretKey\":\"secretValue\"}".bytes.encodeBase64().toString()
+            t.environment 'SECRET_NEW_LINE', "cGFzc3dvcmR2ZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2ZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2\nZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2ZXJ5dmVyeXZleXJsb25ncGFzc3dvcmR2ZXJ5dmVyeXZl\neXJsb25n"
+            t.environment 'SECRET_WEBHOOK_KEY', "secretKey".bytes.encodeBase64().toString()
+            t.environment 'SECRET_NON_B64_SECRET', "some secret value"
+            t.environment 'SECRET_PASSWORD', "cGFzc3dvcmQ="
+            t.environment 'ENV_TEST1', "true"
+            t.environment 'ENV_TEST2', "Pass by env"
 
 
             if (subProj.name == 'core' || subProj.name == 'jdbc-h2' || subProj.name == 'jdbc-mysql' || subProj.name == 'jdbc-postgres') {
                 // JUnit 5 parallel settings
-                systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
-                systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
-                systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
-                systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
+                t.systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
+                t.systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
+                t.systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
+                t.systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
             }
         }
 
+        tasks.register('flakyTest', Test) { Test t ->
+            group = 'verification'
+            description = 'Runs tests tagged @Flaky but does not fail the build.'
+
+            useJUnitPlatform {
+                includeTags 'flaky'
+            }
+            ignoreFailures = true
+
+            reports {
+                junitXml.required = true
+                junitXml.outputPerTestCase = true
+                junitXml.mergeReruns = true
+                junitXml.includeSystemErrLog = true
+                junitXml.outputLocation = layout.buildDirectory.dir("test-results/flakyTest")
+            }
+            commonTestConfig(t)
+
+        }
+
+        test {
+            useJUnitPlatform {
+                excludeTags 'flaky'
+            }
+            reports {
+                junitXml.required = true
+                junitXml.outputPerTestCase = true
+                junitXml.mergeReruns = true
+                junitXml.includeSystemErrLog = true
+                junitXml.outputLocation = layout.buildDirectory.dir("test-results/test")
+            }
+            commonTestConfig(it)
+
+
+            finalizedBy(tasks.named('flakyTest'))
+        }
+
         testlogger {
             theme = 'mocha-parallel'
             showExceptions = true
@@ -344,7 +372,7 @@ tasks.named('testCodeCoverageReport') {
 subprojects {
     sonar {
         properties {
-            property "sonar.coverage.jacoco.xmlReportPaths", "$projectDir.parentFile.path/build/reports/jacoco/testCodeCoverageReport/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/App.java

@@ -43,7 +43,7 @@ import java.util.concurrent.Callable;
         SysCommand.class,
         ConfigCommand.class,
         NamespaceCommand.class,
-        MigrationCommand.class,
+        MigrationCommand.class
     }
 )
 @Introspected

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/sys/SubmitQueuedCommand.java

@@ -6,6 +6,7 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.runners.ExecutionQueued;
+import io.kestra.core.services.ConcurrencyLimitService;
 import io.kestra.jdbc.runner.AbstractJdbcExecutionQueuedStorage;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
@@ -15,8 +16,6 @@ import picocli.CommandLine;
 
 import java.util.Optional;
 
-import static io.kestra.core.utils.Rethrow.throwConsumer;
-
 @CommandLine.Command(
     name = "submit-queued-execution",
     description = {"Submit all queued execution to the executor",
@@ -49,9 +48,11 @@ public class SubmitQueuedCommand extends AbstractCommand {
         }
         else if (queueType.get().equals("postgres") || queueType.get().equals("mysql") || queueType.get().equals("h2")) {
             var executionQueuedStorage = applicationContext.getBean(AbstractJdbcExecutionQueuedStorage.class);
+            var concurrencyLimitService = applicationContext.getBean(ConcurrencyLimitService.class);
 
             for (ExecutionQueued queued : executionQueuedStorage.getAllForAllTenants()) {
-                executionQueuedStorage.pop(queued.getTenantId(), queued.getNamespace(), queued.getFlowId(), throwConsumer(execution -> executionQueue.emit(execution.withState(State.Type.CREATED))));
+                Execution restart = concurrencyLimitService.unqueue(queued.getExecution(), State.Type.RUNNING);
+                executionQueue.emit(restart);
                 cpt++;
             }
         }

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/main/resources/application.yml

@@ -49,6 +49,8 @@ micronaut:
           - /ui/.+
           - /health
           - /health/.+
+          - /metrics
+          - /metrics/.+
           - /prometheus
     http-version: HTTP_1_1
   caches:

cli/src/test/java/io/kestra/cli/commands/configs/sys/NoConfigCommandTest.java

@@ -0,0 +1,76 @@
+package io.kestra.cli.commands.configs.sys;
+import io.kestra.cli.commands.flows.FlowCreateCommand;
+import io.kestra.cli.commands.namespaces.kv.KvCommand;
+import io.micronaut.configuration.picocli.PicocliRunner;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.runtime.server.EmbeddedServer;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Objects;
+
+import static org.assertj.core.api.Assertions.assertThat;
+/**
+ * Verifies CLI behavior without repository configuration:
+ * - Repo-independent commands succeed (e.g. KV with no params).
+ * - Repo-dependent commands fail with a clear error.
+ */
+class NoConfigCommandTest {
+
+    @Test
+    void shouldSucceedWithNamespaceKVCommandWithoutParamsAndConfig() {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.builder().deduceEnvironment(false).start()) {
+            String[] args = {};
+            Integer call = PicocliRunner.call(KvCommand.class, ctx, args);
+
+            assertThat(call).isZero();
+            assertThat(out.toString()).contains("Usage: kestra namespace kv");
+        }
+    }
+
+    @Test
+    void shouldFailWithCreateFlowCommandWithoutConfig() throws URISyntaxException {
+        URL flowUrl = NoConfigCommandTest.class.getClassLoader().getResource("crudFlow/date.yml");
+        Objects.requireNonNull(flowUrl, "Test flow resource not found");
+
+        Path flowPath = Paths.get(flowUrl.toURI());
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        ByteArrayOutputStream err=new ByteArrayOutputStream();
+
+        System.setOut(new PrintStream(out));
+        System.setErr(new PrintStream(err));
+
+        try (ApplicationContext ctx = ApplicationContext.builder()
+            .deduceEnvironment(false)
+            .start()) {
+
+            EmbeddedServer embeddedServer = ctx.getBean(EmbeddedServer.class);
+            embeddedServer.start();
+
+            String[] createArgs = {
+                "--server",
+                embeddedServer.getURL().toString(),
+                "--user",
+                "myuser:pass:word",
+                flowPath.toString(),
+            };
+
+            Integer exitCode = PicocliRunner.call(FlowCreateCommand.class, ctx, createArgs);
+
+
+            assertThat(exitCode).isNotZero();
+            assertThat(out.toString()).isEmpty();
+            assertThat(err.toString()).contains("No bean of type [io.kestra.core.repositories.FlowRepositoryInterface] exists");
+        }
+    }
+
+}

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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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

@@ -91,6 +91,12 @@ public record QueryFilter(
                 return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX, Op.IN, Op.NOT_IN, Op.PREFIX);
             }
         },
+        KIND("kind") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS,Op.NOT_EQUALS);
+            }
+        },
         LABELS("labels") {
             @Override
             public List<Op> supportedOp() {
@@ -211,7 +217,7 @@ public record QueryFilter(
                 return List.of(
                     Field.QUERY, Field.SCOPE, Field.FLOW_ID, Field.START_DATE, Field.END_DATE,
                     Field.STATE, Field.LABELS, Field.TRIGGER_EXECUTION_ID, Field.CHILD_FILTER,
-                    Field.NAMESPACE
+                    Field.NAMESPACE,Field.KIND
                 );
             }
         },
@@ -244,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();
@@ -254,7 +269,7 @@ public record QueryFilter(
          *
          * @return List of {@code ResourceField} with resource names, fields, and operations.
          */
-        
+
         private static FieldOp toFieldInfo(Field field) {
             List<Operation> operations = field.supportedOp().stream()
                 .map(Resource::toOperation)

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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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 = "\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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 = "\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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 = "\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 
     @Valid

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

@@ -496,7 +496,7 @@ public class Execution implements DeletedInterface, TenantInterface {
         }
 
         if (resolvedFinally != null && (
-            this.isTerminated(resolvedTasks, parentTaskRun) || this.hasFailed(resolvedTasks, parentTaskRun
+            this.isTerminated(resolvedTasks, parentTaskRun) || this.hasFailedNoRetry(resolvedTasks, parentTaskRun
         ))) {
             return resolvedFinally;
         }
@@ -584,6 +584,13 @@ public class Execution implements DeletedInterface, TenantInterface {
         );
     }
 
+    public Optional<TaskRun> findLastSubmitted(List<TaskRun> taskRuns) {
+        return Streams.findLast(taskRuns
+            .stream()
+            .filter(t -> t.getState().getCurrent() == State.Type.SUBMITTED)
+        );
+    }
+
     public Optional<TaskRun> findLastRunning(List<TaskRun> taskRuns) {
         return Streams.findLast(taskRuns
             .stream()

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

@@ -61,18 +61,22 @@ public abstract class AbstractFlow implements FlowInterface {
     @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
     @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
     @Schema(
-            description = "Labels as a list of Label (key/value pairs) or as a map of string to string.",
-            oneOf = {
-                    Label[].class,
-                    Map.class
-            }
+        description = "Labels as a list of Label (key/value pairs) or as a map of string to string.",
+        oneOf = {
+            Label[].class,
+            Map.class
+        }
     )
     @Valid
     List<Label> labels;
 
-    @Schema(additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
+    @Schema(
+        type = "object",
+        additionalProperties = Schema.AdditionalPropertiesValue.FALSE
+    )
     Map<String, Object> variables;
 
+
     @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)
@@ -61,6 +61,11 @@ public class Flow extends AbstractFlow implements HasUID {
             }
         });
 
+
+    @Schema(
+        type = "object",
+        additionalProperties = Schema.AdditionalPropertiesValue.FALSE
+    )
     Map<String, Object> variables;
 
     @Valid

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/Input.java

@@ -5,7 +5,6 @@ import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.kestra.core.models.flows.input.*;
 import io.kestra.core.models.property.Property;
-import io.kestra.core.runners.RunContext;
 import io.micronaut.core.annotation.Introspected;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.ConstraintViolationException;
@@ -18,8 +17,6 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
-import java.util.function.Function;
-
 @SuppressWarnings("deprecation")
 @SuperBuilder
 @Getter

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

@@ -1,6 +1,7 @@
 package io.kestra.core.models.flows;
 
 import io.micronaut.core.annotation.Introspected;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.Valid;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -33,6 +34,12 @@ public class Output implements Data {
      * The output value. Can be a dynamic expression.
      */
     @NotNull
+    @Schema(
+        oneOf = {
+            Object.class,
+            String.class
+        }
+    )
     Object value;
 
     /**

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

@@ -2,6 +2,7 @@ package io.kestra.core.models.flows;
 
 import io.kestra.core.validations.PluginDefaultValidation;
 import io.micronaut.core.annotation.Introspected;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
@@ -21,6 +22,10 @@ public class PluginDefault {
     @Builder.Default
     private final boolean forced = false;
 
+    @Schema(
+        type = "object",
+        additionalProperties = Schema.AdditionalPropertiesValue.FALSE
+    )
     private final Map<String, Object> values;
 }
 

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

@@ -222,6 +222,7 @@ public class State {
     @Introspected
     public enum Type {
         CREATED,
+        SUBMITTED,
         RUNNING,
         PAUSED,
         RESTARTED,

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/property/Property.java

@@ -12,6 +12,7 @@ import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
@@ -36,6 +37,12 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor(access = AccessLevel.PACKAGE)
+@Schema(
+    oneOf = {
+        Object.class,
+        String.class
+    }
+)
 public class Property<T> {
     // By default, durations are stored as numbers.
     // We cannot change that globally, as in JDBC/Elastic 'execution.state.duration' must be a number to be able to aggregate them.
@@ -68,7 +75,7 @@ public class Property<T> {
     String getExpression() {
         return expression;
     }
-    
+
     /**
      * Returns a new {@link Property} with no cached rendered value,
      * so that the next render will evaluate its original Pebble expression.
@@ -84,9 +91,9 @@ public class Property<T> {
 
     /**
      * Build a new Property object with a value already set.<br>
-     *
+     * <p>
      * A property build with this method will always return the value passed at build time, no rendering will be done.
-     *
+     * <p>
      * Use {@link #ofExpression(String)} to build a property with a Pebble expression instead.
      */
     public static <V> Property<V> ofValue(V value) {
@@ -126,12 +133,12 @@ public class Property<T> {
 
     /**
      * Build a new Property object with a Pebble expression.<br>
-     *
+     * <p>
      * Use {@link #ofValue(Object)} to build a property with a value instead.
      */
     public static <V> Property<V> ofExpression(@NotNull String expression) {
         Objects.requireNonNull(expression, "'expression' is required");
-        if(!expression.contains("{")) {
+        if (!expression.contains("{")) {
             throw new IllegalArgumentException("'expression' must be a valid Pebble expression");
         }
 
@@ -140,7 +147,7 @@ public class Property<T> {
 
     /**
      * Render a property then convert it to its target type.<br>
-     *
+     * <p>
      * This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
      *
      * @see io.kestra.core.runners.RunContextProperty#as(Class)
@@ -151,14 +158,14 @@ public class Property<T> {
 
     /**
      * Render a property with additional variables, then convert it to its target type.<br>
-     *
+     * <p>
      * This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
      *
      * @see io.kestra.core.runners.RunContextProperty#as(Class, Map)
      */
     public static <T> T as(Property<T> property, PropertyContext context, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
         if (property.value == null) {
-            String rendered =  context.render(property.expression, variables);
+            String rendered = context.render(property.expression, variables);
             property.value = MAPPER.convertValue(rendered, clazz);
         }
 
@@ -167,7 +174,7 @@ public class Property<T> {
 
     /**
      * Render a property then convert it as a list of target type.<br>
-     *
+     * <p>
      * This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
      *
      * @see io.kestra.core.runners.RunContextProperty#asList(Class)
@@ -178,7 +185,7 @@ public class Property<T> {
 
     /**
      * Render a property with additional variables, then convert it as a list of target type.<br>
-     *
+     * <p>
      * This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
      *
      * @see io.kestra.core.runners.RunContextProperty#asList(Class, Map)
@@ -218,25 +225,25 @@ public class Property<T> {
 
     /**
      * Render a property then convert it as a map of target types.<br>
-     *
+     * <p>
      * This method is designed to be used only by the {@link io.kestra.core.runners.RunContextProperty}.
      *
      * @see io.kestra.core.runners.RunContextProperty#asMap(Class, Class)
      */
-    public static <T, K,V> T asMap(Property<T> property, RunContext runContext, Class<K> keyClass, Class<V> valueClass) throws IllegalVariableEvaluationException {
+    public static <T, K, V> T asMap(Property<T> property, RunContext runContext, Class<K> keyClass, Class<V> valueClass) throws IllegalVariableEvaluationException {
         return asMap(property, runContext, keyClass, valueClass, Map.of());
     }
 
     /**
      * Render a property with additional variables, then convert it as a map of target types.<br>
-     *
+     * <p>
      * This method is safe to be used as many times as you want as the rendering and conversion will be cached.
      * Warning, due to the caching mechanism, this method is not thread-safe.
      *
      * @see io.kestra.core.runners.RunContextProperty#asMap(Class, Class, Map)
      */
     @SuppressWarnings({"rawtypes", "unchecked"})
-    public static <T, K,V> T asMap(Property<T> property, RunContext runContext, Class<K> keyClass, Class<V> valueClass, Map<String, Object> variables) throws IllegalVariableEvaluationException {
+    public static <T, K, V> T asMap(Property<T> property, RunContext runContext, Class<K> keyClass, Class<V> valueClass, Map<String, Object> variables) throws IllegalVariableEvaluationException {
         if (property.value == null) {
             JavaType targetMapType = MAPPER.getTypeFactory().constructMapType(Map.class, keyClass, valueClass);
 

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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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/RemoteRunnerInterface.java

@@ -1,3 +1,11 @@
 package io.kestra.core.models.tasks.runners;
 
-public interface RemoteRunnerInterface {}
+import io.kestra.core.models.property.Property;
+import io.swagger.v3.oas.annotations.media.Schema;
+
+public interface RemoteRunnerInterface {
+    @Schema(
+        title = "Whether to synchronize working directory from remote runner back to local one after run."
+    )
+    Property<Boolean> getSyncWorkingDirectory();
+}

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

@@ -30,6 +30,10 @@ public interface TaskCommands {
 
     Map<String, Object> getAdditionalVars();
 
+    default String outputDirectoryName() {
+        return this.getWorkingDirectory().relativize(this.getOutputDirectory()).toString();
+    }
+
     Path getWorkingDirectory();
 
     Path getOutputDirectory();

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

@@ -7,7 +7,6 @@ import io.kestra.core.models.Plugin;
 import io.kestra.core.models.PluginVersioning;
 import io.kestra.core.models.WorkerJobLifecycle;
 import io.kestra.core.models.annotations.PluginProperty;
-import io.kestra.core.models.property.Property;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.core.runner.Process;
 import jakarta.validation.constraints.NotBlank;
@@ -19,13 +18,14 @@ import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 import org.apache.commons.lang3.SystemUtils;
 
-import java.io.IOException;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Stream;
 
 import static io.kestra.core.utils.WindowsUtils.windowsToUnixPath;
+import static io.kestra.core.utils.RegexPatterns.JAVA_IDENTIFIER_REGEX;
 
 /**
  * Base class for all task runners.
@@ -37,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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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,9 +47,9 @@ 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")
     private boolean disabled = false;
 
     @Valid

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/TriggerContext.java

@@ -4,6 +4,7 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.utils.IdUtils;
 import io.micronaut.core.annotation.Introspected;
 import io.micronaut.core.annotation.Nullable;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import lombok.Getter;
@@ -46,6 +47,7 @@ public class TriggerContext {
     @Nullable
     private List<State.Type> stopAfter;
 
+    @Schema(defaultValue = "false")
     private Boolean disabled = Boolean.FALSE;
 
     protected TriggerContext(TriggerContextBuilder<?, ?> b) {

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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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="\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @Pattern(regexp = JAVA_IDENTIFIER_REGEX)
     protected String type;
 }

core/src/main/java/io/kestra/core/queues/QueueFactoryInterface.java

@@ -26,7 +26,6 @@ public interface QueueFactoryInterface {
     String SUBFLOWEXECUTIONRESULT_NAMED = "subflowExecutionResultQueue";
     String CLUSTER_EVENT_NAMED = "clusterEventQueue";
     String SUBFLOWEXECUTIONEND_NAMED = "subflowExecutionEndQueue";
-    String EXECUTION_RUNNING_NAMED = "executionRunningQueue";
     String MULTIPLE_CONDITION_EVENT_NAMED = "multipleConditionEventQueue";
 
     QueueInterface<Execution> execution();
@@ -59,7 +58,5 @@ public interface QueueFactoryInterface {
 
     QueueInterface<SubflowExecutionEnd> subflowExecutionEnd();
 
-    QueueInterface<ExecutionRunning> executionRunning();
-
     QueueInterface<MultipleConditionEvent> multipleConditionEvent();
 }

core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java

@@ -26,8 +26,8 @@ public interface FlowRepositoryInterface extends QueryBuilderInterface<Flows.Fie
      * Used only if result is used internally and not exposed to the user.
      * It is useful when we want to restart/resume a flow.
      */
-    default Flow findByExecutionWithoutAcl(Execution execution) {
-        Optional<Flow> find = this.findByIdWithoutAcl(
+    default FlowWithSource findByExecutionWithoutAcl(Execution execution) {
+        Optional<FlowWithSource> find = this.findByIdWithSourceWithoutAcl(
             execution.getTenantId(),
             execution.getNamespace(),
             execution.getFlowId(),

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

@@ -0,0 +1,31 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.HasUID;
+import io.kestra.core.utils.IdUtils;
+import jakarta.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Value;
+import lombok.With;
+
+@Value
+@AllArgsConstructor
+@Builder
+public class ConcurrencyLimit implements HasUID {
+    @NotNull
+    String tenantId;
+
+    @NotNull
+    String namespace;
+
+    @NotNull
+    String flowId;
+
+    @With
+    Integer running;
+
+    @Override
+    public String uid() {
+        return IdUtils.fromPartsAndSeparator('|', this.tenantId, this.namespace, this.flowId);
+    }
+}

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

@@ -82,6 +82,8 @@ public abstract class FilesService {
     }
 
     private static String resolveUniqueNameForFile(final Path path) {
-        return IdUtils.from(path.toString()) + "-" + path.toFile().getName();
+        String filename = path.getFileName().toString();
+        String encodedFilename = java.net.URLEncoder.encode(filename, java.nio.charset.StandardCharsets.UTF_8);
+        return IdUtils.from(path.toString()) + "-" + encodedFilename;
     }
 }

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

@@ -1,7 +1,6 @@
 package io.kestra.core.runners;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.encryption.EncryptionService;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.KestraRuntimeException;
@@ -65,31 +64,28 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 public class FlowInputOutput {
     private static final Pattern URI_PATTERN = Pattern.compile("^[a-z]+:\\/\\/(?:www\\.)?[-a-zA-Z0-9@:%._\\+~#=]{1,256}\\.[a-zA-Z0-9()]{1,6}\\b(?:[-a-zA-Z0-9()@:%_\\+.~#?&\\/=]*)$");
     private static final ObjectMapper YAML_MAPPER = JacksonMapper.ofYaml();
-
+    
     private final StorageInterface storageInterface;
     private final Optional<String> secretKey;
     private final RunContextFactory runContextFactory;
-    private final VariableRenderer variableRenderer;
-
+    
     @Inject
     public FlowInputOutput(
         StorageInterface storageInterface,
         RunContextFactory runContextFactory,
-        VariableRenderer variableRenderer,
         @Nullable @Value("${kestra.encryption.secret-key}") String secretKey
     ) {
         this.storageInterface = storageInterface;
         this.runContextFactory = runContextFactory;
         this.secretKey = Optional.ofNullable(secretKey);
-        this.variableRenderer = variableRenderer;
     }
-
+    
     /**
      * Validate all the inputs of a given execution of a flow.
      *
-     * @param inputs                  The Flow's inputs.
-     * @param execution               The Execution.
-     * @param data                    The Execution's inputs data.
+     * @param inputs    The Flow's inputs.
+     * @param execution The Execution.
+     * @param data      The Execution's inputs data.
      * @return The list of {@link InputAndValue}.
      */
     public Mono<List<InputAndValue>> validateExecutionInputs(final List<Input<?>> inputs,
@@ -97,10 +93,11 @@ public class FlowInputOutput {
                                                              final Execution execution,
                                                              final Publisher<CompletedPart> data) {
         if (ListUtils.isEmpty(inputs)) return Mono.just(Collections.emptyList());
-
-        return readData(inputs, execution, data, false).map(inputData -> resolveInputs(inputs, flow, execution, inputData));
+        
+        return readData(inputs, execution, data, false)
+            .map(inputData -> resolveInputs(inputs, flow, execution, inputData, false));
     }
-
+    
     /**
      * Reads all the inputs of a given execution of a flow.
      *
@@ -114,7 +111,7 @@ public class FlowInputOutput {
                                                          final Publisher<CompletedPart> data) {
         return this.readExecutionInputs(flow.getInputs(), flow, execution, data);
     }
-
+    
     /**
      * Reads all the inputs of a given execution of a flow.
      *
@@ -129,7 +126,7 @@ public class FlowInputOutput {
                                                          final Publisher<CompletedPart> data) {
         return readData(inputs, execution, data, true).map(inputData -> this.readExecutionInputs(inputs, flow, execution, inputData));
     }
-
+    
     private Mono<Map<String, Object>> readData(List<Input<?>> inputs, Execution execution, Publisher<CompletedPart> data, boolean uploadFiles) {
         return Flux.from(data)
             .publishOn(Schedulers.boundedElastic())
@@ -212,7 +209,7 @@ public class FlowInputOutput {
         final Execution execution,
         final Map<String, ?> data
     ) {
-        Map<String, Object> resolved = this.resolveInputs(inputs, flow, execution, data)
+        Map<String, Object> resolved = this.resolveInputs(inputs, flow, execution, data, true)
             .stream()
             .filter(InputAndValue::enabled)
             .map(it -> {
@@ -238,7 +235,7 @@ public class FlowInputOutput {
         }
         return MapUtils.flattenToNestedMap(resolved);
     }
-
+    
     /**
      * Utility method for retrieving types inputs.
      *
@@ -247,12 +244,21 @@ public class FlowInputOutput {
      * @param data      The Execution's inputs data.
      * @return The Map of typed inputs.
      */
-    @VisibleForTesting
     public List<InputAndValue> resolveInputs(
         final List<Input<?>> inputs,
         final FlowInterface flow,
         final Execution execution,
         final Map<String, ?> data
+    ) {
+        return resolveInputs(inputs, flow, execution, data, true);
+    }
+    
+    public List<InputAndValue> resolveInputs(
+        final List<Input<?>> inputs,
+        final FlowInterface flow,
+        final Execution execution,
+        final Map<String, ?> data,
+        final boolean decryptSecrets
     ) {
         if (inputs == null) {
             return Collections.emptyList();
@@ -262,7 +268,7 @@ public class FlowInputOutput {
             .map(input -> ResolvableInput.of(input,data.get(input.getId())))
             .collect(Collectors.toMap(it -> it.get().input().getId(), Function.identity(), (o1, o2) -> o1, LinkedHashMap::new)));
 
-        resolvableInputMap.values().forEach(input -> resolveInputValue(input, flow, execution, resolvableInputMap));
+        resolvableInputMap.values().forEach(input -> resolveInputValue(input, flow, execution, resolvableInputMap, decryptSecrets));
 
         return resolvableInputMap.values().stream().map(ResolvableInput::get).toList();
     }
@@ -272,7 +278,8 @@ public class FlowInputOutput {
         final @NotNull ResolvableInput resolvable,
         final FlowInterface flow,
         final @NotNull Execution execution,
-        final @NotNull Map<String, ResolvableInput> inputs) {
+        final @NotNull Map<String, ResolvableInput> inputs,
+        final boolean decryptSecrets) {
 
         // return immediately if the input is already resolved
         if (resolvable.isResolved()) return resolvable.get();
@@ -280,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);
-            final RunContext runContext = buildRunContextForExecutionAndInputs(flow, execution, dependencies);
+            // 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);
 
@@ -317,13 +325,13 @@ public class FlowInputOutput {
                 }
             });
             resolvable.setInput(input);
-
-
+            
             Object value = resolvable.get().value();
 
             // 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);
             }
 
@@ -382,7 +390,7 @@ public class FlowInputOutput {
         return Property.asList((Property<List<T>>) input.getDefaults(), renderer, clazz);
     }
 
-    private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies) {
+    private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies, final boolean decryptSecrets) {
         Map<String, Object> flattenInputs = MapUtils.flattenToNestedMap(dependencies.entrySet()
             .stream()
             .collect(HashMap::new, (m, v) -> m.put(v.getKey(), v.getValue().value()), HashMap::putAll)
@@ -396,10 +404,10 @@ public class FlowInputOutput {
                 flattenInputs.put(input.getId(), null);
             }
         }
-        return runContextFactory.of(flow, execution, vars -> vars.withInputs(flattenInputs));
+        return runContextFactory.of(flow, execution, vars -> vars.withInputs(flattenInputs), decryptSecrets);
     }
 
-    private Map<String, InputAndValue> resolveAllDependentInputs(final Input<?> input, final FlowInterface flow, final Execution execution, final Map<String, ResolvableInput> inputs) {
+    private Map<String, InputAndValue> resolveAllDependentInputs(final Input<?> input, final FlowInterface flow, final Execution execution, final Map<String, ResolvableInput> inputs, final boolean decryptSecrets) {
         return Optional.ofNullable(input.getDependsOn())
             .map(DependsOn::inputs)
             .stream()
@@ -407,7 +415,7 @@ public class FlowInputOutput {
             .filter(id -> !id.equals(input.getId()))
             .map(inputs::get)
             .filter(Objects::nonNull) // input may declare unknown or non-necessary dependencies. Let's ignore.
-            .map(it -> resolveInputValue(it, flow, execution, inputs))
+            .map(it -> resolveInputValue(it, flow, execution, inputs, decryptSecrets))
             .collect(Collectors.toMap(it -> it.input().getId(), Function.identity()));
     }
 

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

@@ -84,6 +84,12 @@ public class FlowableUtils {
             return Collections.emptyList();
         }
 
+        // have submitted, leave
+        Optional<TaskRun> lastSubmitted = execution.findLastSubmitted(taskRuns);
+        if (lastSubmitted.isPresent()) {
+            return Collections.emptyList();
+        }
+
         // have running, leave
         Optional<TaskRun> lastRunning = execution.findLastRunning(taskRuns);
         if (lastRunning.isPresent()) {

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

@@ -41,6 +41,9 @@ public class RunContextFactory {
 
     @Inject
     protected VariableRenderer variableRenderer;
+    
+    @Inject
+    protected SecureVariableRendererFactory secureVariableRendererFactory;
 
     @Inject
     protected StorageInterface storageInterface;
@@ -82,22 +85,33 @@ public class RunContextFactory {
     public RunContext of(FlowInterface flow, Execution execution) {
         return of(flow, execution, Function.identity());
     }
+    
+    public RunContext of(FlowInterface flow, Execution execution, boolean decryptVariable) {
+        return of(flow, execution, Function.identity(), decryptVariable);
+    }
 
     public RunContext of(FlowInterface flow, Execution execution, Function<RunVariables.Builder, RunVariables.Builder> runVariableModifier) {
+        return of(flow, execution, runVariableModifier, true);
+    }
+    
+    public RunContext of(FlowInterface flow, Execution execution, Function<RunVariables.Builder, RunVariables.Builder> runVariableModifier, boolean decryptVariables) {
         RunContextLogger runContextLogger = runContextLoggerFactory.create(execution);
-
+        
+        VariableRenderer variableRenderer = decryptVariables ? this.variableRenderer : secureVariableRendererFactory.createOrGet();
+        
         return newBuilder()
             // Logger
             .withLogger(runContextLogger)
             // Execution
             .withPluginConfiguration(Map.of())
             .withStorage(new InternalStorage(runContextLogger.logger(), StorageContext.forExecution(execution), storageInterface, flowService))
+            .withVariableRenderer(variableRenderer)
             .withVariables(runVariableModifier.apply(
-                newRunVariablesBuilder()
-                    .withFlow(flow)
-                    .withExecution(execution)
-                    .withDecryptVariables(true)
-                    .withSecretInputs(secretInputsFromFlow(flow))
+                    newRunVariablesBuilder()
+                        .withFlow(flow)
+                        .withExecution(execution)
+                        .withDecryptVariables(decryptVariables)
+                        .withSecretInputs(secretInputsFromFlow(flow))
                 )
                 .build(runContextLogger, PropertyContext.create(variableRenderer)))
             .withSecretInputs(secretInputsFromFlow(flow))
@@ -109,7 +123,7 @@ public class RunContextFactory {
     }
 
     public RunContext of(FlowInterface flow, Task task, Execution execution, TaskRun taskRun, boolean decryptVariables) {
-        return this.of(flow, task, execution, taskRun, decryptVariables, variableRenderer);
+        return this.of(flow, task, execution, taskRun, decryptVariables, this.variableRenderer);
     }
 
     public RunContext of(FlowInterface flow, Task task, Execution execution, TaskRun taskRun, boolean decryptVariables, VariableRenderer variableRenderer) {
@@ -147,7 +161,7 @@ public class RunContextFactory {
                 .withFlow(flow)
                 .withTrigger(trigger)
                 .withSecretInputs(secretInputsFromFlow(flow))
-                .build(runContextLogger, PropertyContext.create(variableRenderer))
+                .build(runContextLogger, PropertyContext.create(this.variableRenderer))
             )
             .withSecretInputs(secretInputsFromFlow(flow))
             .withTrigger(trigger)
@@ -226,7 +240,7 @@ public class RunContextFactory {
             // inject mandatory services and config
             .withApplicationContext(applicationContext) // TODO - ideally application should not be injected here
             .withMeterRegistry(metricRegistry)
-            .withVariableRenderer(variableRenderer)
+            .withVariableRenderer(this.variableRenderer)
             .withStorageInterface(storageInterface)
             .withSecretKey(secretKey)
             .withWorkingDir(workingDirFactory.createWorkingDirectory())

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

@@ -0,0 +1,39 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.runners.pebble.PebbleEngineFactory;
+import io.kestra.core.runners.pebble.functions.SecretFunction;
+import io.micronaut.context.ApplicationContext;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import java.util.List;
+
+@Singleton
+public class SecureVariableRendererFactory {
+    
+    private final PebbleEngineFactory pebbleEngineFactory;
+    private final ApplicationContext applicationContext;
+    
+    private VariableRenderer secureVariableRenderer;
+    
+    @Inject
+    public SecureVariableRendererFactory(ApplicationContext applicationContext, PebbleEngineFactory pebbleEngineFactory) {
+        this.pebbleEngineFactory = pebbleEngineFactory;
+        this.applicationContext = applicationContext;
+    }
+    
+    /**
+     * Creates or returns the existing secured {@link VariableRenderer} instance.
+     *
+     * @return the secured {@link VariableRenderer} instance
+     */
+    public synchronized VariableRenderer createOrGet() {
+        if (this.secureVariableRenderer == null) {
+            // Explicitly create a new instance through the application context to ensure
+            // eventual custom VariableRenderer implementation is used
+            secureVariableRenderer = applicationContext.createBean(VariableRenderer.class);
+            secureVariableRenderer.setPebbleEngine(pebbleEngineFactory.createWithMaskedFunctions(secureVariableRenderer, List.of(SecretFunction.NAME)));
+        }
+        return secureVariableRenderer;
+    }
+}
+

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

@@ -2,121 +2,44 @@ package io.kestra.core.runners;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.runners.pebble.*;
-import io.kestra.core.runners.pebble.functions.RenderingFunctionInterface;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.annotation.ConfigurationProperties;
 import io.micronaut.core.annotation.Nullable;
 import io.pebbletemplates.pebble.PebbleEngine;
 import io.pebbletemplates.pebble.error.AttributeNotFoundException;
 import io.pebbletemplates.pebble.error.PebbleException;
-import io.pebbletemplates.pebble.extension.Extension;
-import io.pebbletemplates.pebble.extension.Function;
 import io.pebbletemplates.pebble.template.PebbleTemplate;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import lombok.Getter;
 
 import java.io.IOException;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Proxy;
 import java.util.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 
 @Singleton
 public class VariableRenderer {
     private static final Pattern RAW_PATTERN = Pattern.compile("(\\{%-*\\s*raw\\s*-*%}(.*?)\\{%-*\\s*endraw\\s*-*%})");
     public static final int MAX_RENDERING_AMOUNT = 100;
 
-    private final PebbleEngine pebbleEngine;
+    private PebbleEngine pebbleEngine;
     private final VariableConfiguration variableConfiguration;
 
     @Inject
     public VariableRenderer(ApplicationContext applicationContext, @Nullable VariableConfiguration variableConfiguration) {
-        this(applicationContext, variableConfiguration, Collections.emptyList());
+        this(applicationContext.getBean(PebbleEngineFactory.class), variableConfiguration);
     }
-
-    public VariableRenderer(ApplicationContext applicationContext, @Nullable VariableConfiguration variableConfiguration, List<String> functionsToMask) {
+    
+    public VariableRenderer(PebbleEngineFactory pebbleEngineFactory, @Nullable VariableConfiguration variableConfiguration) {
         this.variableConfiguration = variableConfiguration != null ? variableConfiguration : new VariableConfiguration();
-
-        PebbleEngine.Builder pebbleBuilder = new PebbleEngine.Builder()
-            .registerExtensionCustomizer(ExtensionCustomizer::new)
-            .strictVariables(true)
-            .cacheActive(this.variableConfiguration.getCacheEnabled())
-            .newLineTrimming(false)
-            .autoEscaping(false);
-
-        List<Extension> extensions = applicationContext.getBeansOfType(Extension.class).stream()
-            .map(e -> functionsToMask.stream().anyMatch(excludedFunction -> e.getFunctions().containsKey(excludedFunction))
-                ? extensionWithMaskedFunctions(e, functionsToMask)
-                : e)
-            .toList();
-
-        extensions.forEach(pebbleBuilder::extension);
-
-        if (this.variableConfiguration.getCacheEnabled()) {
-            pebbleBuilder.templateCache(new PebbleLruCache(this.variableConfiguration.getCacheSize()));
-        }
-
-        this.pebbleEngine = pebbleBuilder.build();
+        this.pebbleEngine = pebbleEngineFactory.create();
     }
-
-    private Extension extensionWithMaskedFunctions(Extension initialExtension, List<String> maskedFunctions) {
-        return (Extension) Proxy.newProxyInstance(
-            initialExtension.getClass().getClassLoader(),
-            new Class[]{Extension.class},
-            (proxy, method, methodArgs) -> {
-                if (method.getName().equals("getFunctions")) {
-                    return initialExtension.getFunctions().entrySet().stream()
-                        .map(entry -> {
-                            if (maskedFunctions.contains(entry.getKey())) {
-                                return Map.entry(entry.getKey(), this.maskedFunctionProxy(entry.getValue()));
-                            } else if (RenderingFunctionInterface.class.isAssignableFrom(entry.getValue().getClass())) {
-                                return Map.entry(entry.getKey(), this.variableRendererProxy(entry.getValue()));
-                            }
-
-                            return entry;
-                        }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-                }
-
-                return method.invoke(initialExtension, methodArgs);
-            }
-        );
+    
+    public void setPebbleEngine(final PebbleEngine pebbleEngine) {
+        this.pebbleEngine = pebbleEngine;
     }
-
-    private Function variableRendererProxy(Function initialFunction) {
-        return (Function) Proxy.newProxyInstance(
-            initialFunction.getClass().getClassLoader(),
-            new Class[]{Function.class, RenderingFunctionInterface.class},
-            (functionProxy, functionMethod, functionArgs) -> {
-                if (functionMethod.getName().equals("variableRenderer")) {
-                    return this;
-                }
-                return functionMethod.invoke(initialFunction, functionArgs);
-            }
-        );
-    }
-
-    private Function maskedFunctionProxy(Function initialFunction) {
-        return (Function) Proxy.newProxyInstance(
-            initialFunction.getClass().getClassLoader(),
-            new Class[]{Function.class},
-            (functionProxy, functionMethod, functionArgs) -> {
-                Object result;
-                try {
-                    result = functionMethod.invoke(initialFunction, functionArgs);
-                } catch (InvocationTargetException e) {
-                    throw e.getCause();
-                }
-                if (functionMethod.getName().equals("execute")) {
-                    return "******";
-                }
-                return result;
-            }
-        );
-    }
-
+    
     public static IllegalVariableEvaluationException properPebbleException(PebbleException initialExtension) {
         if (initialExtension instanceof AttributeNotFoundException current) {
             return new IllegalVariableEvaluationException(

core/src/main/java/io/kestra/core/runners/pebble/PebbleEngineFactory.java

@@ -0,0 +1,118 @@
+package io.kestra.core.runners.pebble;
+
+import io.kestra.core.runners.VariableRenderer;
+import io.kestra.core.runners.pebble.functions.RenderingFunctionInterface;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.core.annotation.Nullable;
+import io.pebbletemplates.pebble.PebbleEngine;
+import io.pebbletemplates.pebble.extension.Extension;
+import io.pebbletemplates.pebble.extension.Function;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Singleton
+public class PebbleEngineFactory {
+    
+    private final ApplicationContext applicationContext;
+    private final VariableRenderer.VariableConfiguration variableConfiguration;
+    
+    @Inject
+    public PebbleEngineFactory(ApplicationContext applicationContext, @Nullable VariableRenderer.VariableConfiguration variableConfiguration) {
+        this.applicationContext = applicationContext;
+        this.variableConfiguration = variableConfiguration;
+    }
+    
+    public PebbleEngine create() {
+        PebbleEngine.Builder builder = newPebbleEngineBuilder();
+        this.applicationContext.getBeansOfType(Extension.class).forEach(builder::extension);
+        return builder.build();
+    }
+    
+    public PebbleEngine createWithMaskedFunctions(VariableRenderer renderer, final List<String> functionsToMask) {
+        
+        PebbleEngine.Builder builder = newPebbleEngineBuilder();
+        
+        this.applicationContext.getBeansOfType(Extension.class).stream()
+            .map(e -> functionsToMask.stream().anyMatch(fun -> e.getFunctions().containsKey(fun))
+                ? extensionWithMaskedFunctions(renderer, e, functionsToMask)
+                : e)
+            .forEach(builder::extension);
+        
+        return builder.build();
+    }
+    
+    private PebbleEngine.Builder newPebbleEngineBuilder() {
+        PebbleEngine.Builder builder = new PebbleEngine.Builder()
+            .registerExtensionCustomizer(ExtensionCustomizer::new)
+            .strictVariables(true)
+            .cacheActive(this.variableConfiguration.getCacheEnabled())
+            .newLineTrimming(false)
+            .autoEscaping(false);
+        
+        if (this.variableConfiguration.getCacheEnabled()) {
+            builder = builder.templateCache(new PebbleLruCache(this.variableConfiguration.getCacheSize()));
+        }
+        return builder;
+    }
+    
+    private Extension extensionWithMaskedFunctions(VariableRenderer renderer, Extension initialExtension, List<String> maskedFunctions) {
+        return (Extension) Proxy.newProxyInstance(
+            initialExtension.getClass().getClassLoader(),
+            new Class[]{Extension.class},
+            (proxy, method, methodArgs) -> {
+                if (method.getName().equals("getFunctions")) {
+                    return initialExtension.getFunctions().entrySet().stream()
+                        .map(entry -> {
+                            if (maskedFunctions.contains(entry.getKey())) {
+                                return Map.entry(entry.getKey(), this.maskedFunctionProxy(entry.getValue()));
+                            } else if (RenderingFunctionInterface.class.isAssignableFrom(entry.getValue().getClass())) {
+                                return Map.entry(entry.getKey(), this.variableRendererProxy(renderer, entry.getValue()));
+                            }
+                            
+                            return entry;
+                        }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+                }
+                
+                return method.invoke(initialExtension, methodArgs);
+            }
+        );
+    }
+    
+    private Function variableRendererProxy(VariableRenderer renderer, Function initialFunction) {
+        return (Function) Proxy.newProxyInstance(
+            initialFunction.getClass().getClassLoader(),
+            new Class[]{Function.class, RenderingFunctionInterface.class},
+            (functionProxy, functionMethod, functionArgs) -> {
+                if (functionMethod.getName().equals("variableRenderer")) {
+                    return renderer;
+                }
+                return functionMethod.invoke(initialFunction, functionArgs);
+            }
+        );
+    }
+    
+    private Function maskedFunctionProxy(Function initialFunction) {
+        return (Function) Proxy.newProxyInstance(
+            initialFunction.getClass().getClassLoader(),
+            new Class[]{Function.class},
+            (functionProxy, functionMethod, functionArgs) -> {
+                Object result;
+                try {
+                    result = functionMethod.invoke(initialFunction, functionArgs);
+                } catch (InvocationTargetException e) {
+                    throw e.getCause();
+                }
+                if (functionMethod.getName().equals("execute")) {
+                    return "******";
+                }
+                return result;
+            }
+        );
+    }
+}

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/services/StorageService.java

@@ -14,8 +14,12 @@ import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.BiFunction;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import static io.kestra.core.utils.Rethrow.throwConsumer;
 import static io.kestra.core.utils.Rethrow.throwFunction;
@@ -32,7 +36,11 @@ public abstract class StorageService {
         try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(runContext.storage().getFile(from)))) {
             List<Path> splited;
 
-            if (storageSplitInterface.getBytes() != null) {
+            if (storageSplitInterface.getRegexPattern() != null) {
+                String renderedPattern = runContext.render(storageSplitInterface.getRegexPattern()).as(String.class).orElseThrow();
+                String separator = runContext.render(storageSplitInterface.getSeparator()).as(String.class).orElseThrow();
+                splited = StorageService.splitByRegex(runContext, extension, separator, bufferedReader, renderedPattern);
+            } else if (storageSplitInterface.getBytes() != null) {
                 ReadableBytesTypeConverter readableBytesTypeConverter = new ReadableBytesTypeConverter();
                 Number convert = readableBytesTypeConverter.convert(runContext.render(storageSplitInterface.getBytes()).as(String.class).orElseThrow(), Number.class)
                     .orElseThrow(() -> new IllegalArgumentException("Invalid size with value '" + storageSplitInterface.getBytes() + "'"));
@@ -47,7 +55,7 @@ public abstract class StorageService {
                 splited = StorageService.split(runContext, extension, runContext.render(storageSplitInterface.getSeparator()).as(String.class).orElseThrow(),
                     bufferedReader, (bytes, size) -> size >= renderedRows);
             } else {
-                throw new IllegalArgumentException("Invalid configuration with no size, count, nor rows");
+                throw new IllegalArgumentException("Invalid configuration with no size, count, rows, nor regexPattern");
             }
 
             return splited
@@ -117,4 +125,36 @@ public abstract class StorageService {
         return files.stream().filter(p -> p.toFile().length() > 0).toList();
     }
 
+    private static List<Path> splitByRegex(RunContext runContext, String extension, String separator, BufferedReader bufferedReader, String regexPattern) throws IOException {
+        List<Path> files = new ArrayList<>();
+        Map<String, RandomAccessFile> writers = new HashMap<>();
+        Pattern pattern = Pattern.compile(regexPattern);
+        
+        String row;
+        while ((row = bufferedReader.readLine()) != null) {
+            Matcher matcher = pattern.matcher(row);
+            
+            if (matcher.find() && matcher.groupCount() > 0) {
+                String routingKey = matcher.group(1);
+                
+                // Get or create writer for this routing key
+                RandomAccessFile writer = writers.get(routingKey);
+                if (writer == null) {
+                    Path path = runContext.workingDir().createTempFile(extension);
+                    files.add(path);
+                    writer = new RandomAccessFile(path.toFile(), "rw");
+                    writers.put(routingKey, writer);
+                }
+                
+                byte[] bytes = (row + separator).getBytes(StandardCharsets.UTF_8);
+                writer.getChannel().write(ByteBuffer.wrap(bytes));
+            }
+            // Lines that don't match the pattern are ignored
+        }
+        
+        writers.values().forEach(throwConsumer(RandomAccessFile::close));
+        
+        return files.stream().filter(p -> p.toFile().length() > 0).toList();
+    }
+
 }

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/storages/StorageSplitInterface.java

@@ -26,4 +26,13 @@ public interface StorageSplitInterface {
         defaultValue = "\\n"
     )
     Property<String> getSeparator();
+
+    @Schema(
+        title = "Split file by regex pattern with the first capture group as the routing key.",
+        description = "A regular expression pattern with a capture group. Lines matching this pattern will be grouped by the captured value. " +
+            "For example, `\\[(\\w+)\\]` will group lines by log level (ERROR, WARN, INFO) extracted from log entries. " +
+            "Lines with the same captured value will be written to the same output file. " +
+            "This allows content-based splitting where the file is divided based on data patterns rather than size or line count."
+    )
+    Property<String> getRegexPattern();
 }

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/core/utils/TruthUtils.java

@@ -6,10 +6,10 @@ abstract public class TruthUtils {
     private static final List<String> FALSE_VALUES = List.of("false", "0", "-0", "");
 
     public static boolean isTruthy(String condition) {
-        return condition != null && !FALSE_VALUES.contains(condition);
+        return condition != null && !FALSE_VALUES.contains(condition.trim());
     }
 
     public static boolean isFalsy(String condition) {
-        return condition != null && FALSE_VALUES.contains(condition);
+        return condition != null && FALSE_VALUES.contains(condition.trim());
     }
 }

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 = "\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    @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/flow/ForEachItem.java

@@ -648,6 +648,8 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
 
         @Builder.Default
         private Property<String> separator = Property.ofValue("\n");
+
+        private Property<String> regexPattern;
     }
 
     @Builder

core/src/main/java/io/kestra/plugin/core/http/Download.java

@@ -173,8 +173,8 @@ public class Download extends AbstractHttp implements RunnableTask<Download.Outp
         if (path.indexOf('/') != -1) {
             path = path.substring(path.lastIndexOf('/')); // keep the last segment
         }
-        if (path.indexOf('.') != -1) {
-            return path.substring(path.indexOf('.'));
+        if (path.lastIndexOf('.') != -1) {
+            return path.substring(path.lastIndexOf('.'));
         }
         return null;
     }

core/src/main/java/io/kestra/plugin/core/storage/Split.java

@@ -48,6 +48,13 @@ import java.util.List;
                 "partitions: 8"
             }
         ),
+        @Example(
+            title = "Split a file by regex pattern - group lines by captured value.",
+            code = {
+                "from: \"kestra://long/url/logs.txt\"",
+                "regexPattern: \"\\\\[(\\\\w+)\\\\]\""
+            }
+        ),
     },
     aliases = "io.kestra.core.tasks.storages.Split"
 )
@@ -65,6 +72,13 @@ public class Split extends Task implements RunnableTask<Split.Output>, StorageSp
 
     private Property<Integer> rows;
 
+    @Schema(
+        title = "Split file by regex pattern. Lines are grouped by the first capture group value.",
+        description = "A regular expression pattern with a capture group. Lines matching this pattern will be grouped by the captured value. For example, `\\[(\\w+)\\]` will group lines by log level (ERROR, WARN, INFO) extracted from log entries."
+    )
+    @PluginProperty(dynamic = true)
+    private Property<String> regexPattern;
+
     @Builder.Default
     private Property<String> separator = Property.ofValue("\n");
 

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?