fix(core): fix search lookup for endDate field

Co-authored-by: Loïc Mathieu <loikeseke@gmail.com>

.devcontainer/README.md

@@ -27,11 +27,6 @@ In the meantime, you can move onto the next step...
 
 - Create a `.env.development.local` file in the `ui` folder and paste the following:
 
-```bash
-# This lets the frontend know what the backend URL is but you are free to change this to your actual server URL e.g. hosted version of Kestra.
-VITE_APP_API_URL=http://localhost:8080
-```
-
 - Navigate into the `ui` folder and run `npm install` to install the dependencies for the frontend project.
 
 - Now go to the `cli/src/main/resources` folder and create a `application-override.yml` file.
@@ -74,9 +69,6 @@ kestra:
       path: /tmp/kestra-wd/tmp
   anonymous-usage-report:
     enabled: false
-  server:
-    basic-auth:
-      enabled: false
 
 datasources:
   postgres:

.github/CONTRIBUTING.md

@@ -80,7 +80,6 @@ python3 -m pip install virtualenv
 The frontend is made with [Vue.js](https://vuejs.org/) and located on the `/ui` folder.
 
 - `npm install`
-- create a file `ui/.env.development.local` with content `VITE_APP_API_URL=http://localhost:8080` (or your actual server url)
 - `npm run dev` will start the development server with hot reload.
 - The server start by default on port 5173 and is reachable on `http://localhost:5173`
 - You can run `npm run build` in order to build the front-end that will be delivered from the backend (without running the `npm run dev`) above.

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

@@ -43,9 +43,6 @@ jobs:
           with:
             node-version: "20.x"
 
-        - name: Check keys matching
-          run: node ui/src/translations/check.js
-
         - name: Set up Git
           run: |
             git config --global user.name "GitHub Action"
@@ -64,4 +61,7 @@ jobs:
             fi
             git commit -m "chore(core): localize to languages other than english" -m "Extended localization support by adding translations for multiple languages using English as the base. This enhances accessibility and usability for non-English-speaking users while keeping English as the source reference."
             git push -u origin $BRANCH_NAME || (git push origin --delete $BRANCH_NAME && git push -u origin $BRANCH_NAME)
-            gh pr create --title "Translations from en.json" --body "This PR was created automatically by a GitHub Action." --base develop --head $BRANCH_NAME --assignee anna-geller --reviewer anna-geller
+            gh pr create --title "Translations from en.json" --body $'This PR was created automatically by a GitHub Action.\n\nSomeone from the @kestra-io/frontend team needs to review and merge.' --base ${{ github.ref_name }} --head $BRANCH_NAME
+
+        - name: Check keys matching
+          run: node ui/src/translations/check.js

.github/workflows/docker.yml

@@ -7,7 +7,7 @@ on:
         description: 'Retag latest Docker images'
         required: true
         type: string
-        default: "true"
+        default: "false"
         options:
           - "true"
           - "false"
@@ -69,7 +69,7 @@ jobs:
           fi
 
           if [[ "${{ env.PLUGIN_VERSION }}" == *"-SNAPSHOT" ]]; then
-            echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots ${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT;
+            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ ${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT;
           else
             echo "plugins=${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT
           fi

.github/workflows/e2e.yml

@@ -36,6 +36,15 @@ jobs:
         with:
           path: kestra
 
+      # Setup build
+      - uses: kestra-io/actions/.github/actions/setup-build@main
+        name: Setup - Build
+        id: build
+        with:
+          java-enabled: true
+          node-enabled: true
+          python-enabled: true
+
       - name: Install Npm dependencies
         run: |
           cd kestra/ui
@@ -44,8 +53,8 @@ jobs:
 
       - name: Run E2E Tests
         run: |
-          cd kestra/ui
-          npm run test:e2e
+          cd kestra
+          sh build-and-start-e2e-tests.sh
 
       - name: Upload Playwright Report as Github artifact
         # 'With this report, you can analyze locally the results of the tests. see https://playwright.dev/docs/ci-intro#html-report'
@@ -53,7 +62,7 @@ jobs:
         if: ${{ !cancelled() }}
         with:
           name: playwright-report
-          path: kestra/playwright-report/
+          path: kestra/ui/playwright-report/
           retention-days: 7
         # Allure check
         # TODO I don't know what it should do
@@ -74,4 +83,4 @@ jobs:
 #          baseUrl: "https://internal.dev.kestra.io"
 #          prefix: ${{ format('{0}/{1}', github.repository, 'allure/java') }}
 #          copyLatest: true
-#          ignoreMissingResults: true
+#          ignoreMissingResults: true

.github/workflows/pull-request.yml

@@ -56,6 +56,10 @@ jobs:
       SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
       GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
 
+  e2e-tests:
+    name: E2E - Tests
+    uses: ./.github/workflows/e2e.yml
+
   end:
     name: End
     runs-on: ubuntu-latest

.github/workflows/vulnerabilities-check.yml

@@ -87,7 +87,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.31.0
+        uses: aquasecurity/trivy-action@0.32.0
         with:
           image-ref: kestra/kestra:develop
           format: 'template'
@@ -132,7 +132,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.31.0
+        uses: aquasecurity/trivy-action@0.32.0
         with:
           image-ref: kestra/kestra:latest
           format: table

.github/workflows/workflow-build-artifacts.yml

@@ -68,7 +68,7 @@ jobs:
           if [[ $TAG = "master" || $TAG == v* ]]; then
             echo "plugins=$PLUGINS" >> $GITHUB_OUTPUT
           else
-            echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots $PLUGINS" >> $GITHUB_OUTPUT
+            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
           fi
 
       # Build

.github/workflows/workflow-frontend-test.yml

@@ -39,7 +39,6 @@ jobs:
           key: playwright-${{ hashFiles('ui/package-lock.json') }}
 
       - name: Npm - install
-        shell: bash
         if: steps.cache-node-modules.outputs.cache-hit != 'true'
         working-directory: ui
         run: npm ci
@@ -52,35 +51,20 @@ jobs:
           workdir: ui
 
       - name: Npm - Run build
-        shell: bash
         working-directory: ui
         env:
           CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
         run: npm run build
 
+      - name: Run front-end unit tests
+        working-directory: ui
+        run: npm run test:unit -- --coverage
+
       - name: Storybook - Install Playwright
-        shell: bash
         working-directory: ui
         if: steps.cache-playwright.outputs.cache-hit != 'true'
         run: npx playwright install --with-deps
 
-      - name: Run front-end unit tests
-        shell: bash
+      - name: Run storybook component tests
         working-directory: ui
-        run: npm run test:cicd
-
-      - name: Codecov - Upload coverage reports
-        uses: codecov/codecov-action@v5
-        if: ${{ !cancelled() && github.event.pull_request.head.repo.full_name == github.repository }}
-        continue-on-error: true
-        with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          flags: frontend
-
-      - name: Codecov - Upload test results
-        uses: codecov/test-results-action@v1
-        if: ${{ !cancelled() }}
-        continue-on-error: true
-        with:
-          token: ${{ secrets.CODECOV_TOKEN && github.event.pull_request.head.repo.full_name == github.repository }}
-          flags: frontend
+        run: npm run test:storybook -- --coverage

.github/workflows/workflow-github-release.yml

@@ -41,12 +41,25 @@ jobs:
           name: exe
           path: build/executable
 
+      - name: Check if current tag is latest
+        id: is_latest
+        run: |
+          latest_tag=$(git tag | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | sed 's/^v//' | sort -V | tail -n1)
+          current_tag="${GITHUB_REF_NAME#v}"
+          if [ "$current_tag" = "$latest_tag" ]; then
+          echo "latest=true" >> $GITHUB_OUTPUT
+          else
+          echo "latest=false" >> $GITHUB_OUTPUT
+          fi
+        env:
+          GITHUB_REF_NAME: ${{ github.ref_name }}
 
       # GitHub Release
       - name: Create GitHub release
         uses: ./actions/.github/actions/github-release
         if: ${{ startsWith(github.ref, 'refs/tags/v') }}
         env:
+          MAKE_LATEST: ${{ steps.is_latest.outputs.latest }}
           GITHUB_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
           SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
 

.github/workflows/workflow-publish-docker.yml

@@ -112,12 +112,12 @@ jobs:
             echo "plugins=${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT
           elif [[ $TAG = "develop" ]]; then
             TAG="develop";
-            echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots $PLUGINS" >> $GITHUB_OUTPUT
+            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
           else
             TAG="build-${{ github.run_id }}";
-            echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots $PLUGINS" >> $GITHUB_OUTPUT
+            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
           fi
-          
+
           echo "tag=${TAG}${{ matrix.image.tag }}" >> $GITHUB_OUTPUT
 
       # Build Docker Image

.github/workflows/workflow-publish-maven.yml

@@ -39,8 +39,8 @@ jobs:
       - name: Publish - Release package to Maven Central
         shell: bash
         env:
-          ORG_GRADLE_PROJECT_sonatypeUsername: ${{ secrets.SONATYPE_USER }}
-          ORG_GRADLE_PROJECT_sonatypePassword: ${{ secrets.SONATYPE_PASSWORD }}
+          ORG_GRADLE_PROJECT_mavenCentralUsername: ${{ secrets.SONATYPE_USER }}
+          ORG_GRADLE_PROJECT_mavenCentralPassword: ${{ secrets.SONATYPE_PASSWORD }}
           SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
           SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
           SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE}}
@@ -50,7 +50,7 @@ jobs:
           echo "signing.password=${SONATYPE_GPG_PASSWORD}" >> ~/.gradle/gradle.properties
           echo "signing.secretKeyRingFile=${HOME}/.gradle/secring.gpg" >> ~/.gradle/gradle.properties
           echo ${SONATYPE_GPG_FILE} | base64 -d > ~/.gradle/secring.gpg
-          ./gradlew publishToSonatype ${{ startsWith(github.ref, 'refs/tags/v')  && 'closeAndReleaseSonatypeStagingRepository' || '' }}
+          ./gradlew publishToMavenCentral
 
       # Gradle dependency
       - name: Java - Gradle dependency graph

.plugins

@@ -3,10 +3,12 @@
 # Format: <RepositoryName>:<GroupId>:<ArtifactId>:<Version>
 #
 # Uncomment the lines corresponding to the plugins to be installed:
+#plugin-ai:io.kestra.plugin:plugin-ai:LATEST
 #plugin-airbyte:io.kestra.plugin:plugin-airbyte:LATEST
 #plugin-airflow:io.kestra.plugin:plugin-airflow:LATEST
 #plugin-amqp:io.kestra.plugin:plugin-amqp:LATEST
 #plugin-ansible:io.kestra.plugin:plugin-ansible:LATEST
+#plugin-anthropic:io.kestra.plugin:plugin-anthropic:LATEST
 #plugin-aws:io.kestra.plugin:plugin-aws:LATEST
 #plugin-azure:io.kestra.plugin:plugin-azure:LATEST
 #plugin-cassandra:io.kestra.plugin:plugin-cassandra:LATEST
@@ -29,6 +31,7 @@
 #plugin-fivetran:io.kestra.plugin:plugin-fivetran:LATEST
 #plugin-fs:io.kestra.plugin:plugin-fs:LATEST
 #plugin-gcp:io.kestra.plugin:plugin-gcp:LATEST
+#plugin-gemini:io.kestra.plugin:plugin-gemini:LATEST
 #plugin-git:io.kestra.plugin:plugin-git:LATEST
 #plugin-github:io.kestra.plugin:plugin-github:LATEST
 #plugin-googleworkspace:io.kestra.plugin:plugin-googleworkspace:LATEST
@@ -63,21 +66,23 @@
 #plugin-kafka:io.kestra.plugin:plugin-kafka:LATEST
 #plugin-kestra:io.kestra.plugin:plugin-kestra:LATEST
 #plugin-kubernetes:io.kestra.plugin:plugin-kubernetes:LATEST
-#plugin-langchain4j:io.kestra.plugin:plugin-langchain4j:LATEST
 #plugin-ldap:io.kestra.plugin:plugin-ldap:LATEST
 #plugin-linear:io.kestra.plugin:plugin-linear:LATEST
 #plugin-malloy:io.kestra.plugin:plugin-malloy:LATEST
 #plugin-meilisearch:io.kestra.plugin:plugin-meilisearch:LATEST
 #plugin-minio:io.kestra.plugin:plugin-minio:LATEST
+#plugin-mistral:io.kestra.plugin:plugin-mistral:LATEST
 #plugin-modal:io.kestra.plugin:plugin-modal:LATEST
 #plugin-mongodb:io.kestra.plugin:plugin-mongodb:LATEST
 #plugin-mqtt:io.kestra.plugin:plugin-mqtt:LATEST
 #plugin-nats:io.kestra.plugin:plugin-nats:LATEST
 #plugin-neo4j:io.kestra.plugin:plugin-neo4j:LATEST
 #plugin-notifications:io.kestra.plugin:plugin-notifications:LATEST
+#plugin-notion:io.kestra.plugin:plugin-notion:LATEST
 #plugin-ollama:io.kestra.plugin:plugin-ollama:LATEST
 #plugin-openai:io.kestra.plugin:plugin-openai:LATEST
 #plugin-opensearch:io.kestra.plugin:plugin-opensearch:LATEST
+#plugin-perplexity:io.kestra.plugin:plugin-perplexity:LATEST
 #plugin-powerbi:io.kestra.plugin:plugin-powerbi:LATEST
 #plugin-pulsar:io.kestra.plugin:plugin-pulsar:LATEST
 #plugin-redis:io.kestra.plugin:plugin-redis:LATEST

Makefile

@@ -77,7 +77,7 @@ install-plugins:
         else \
 		${KESTRA_BASEDIR}/bin/kestra plugins install $$CURRENT_PLUGIN \
 		--plugins ${KESTRA_BASEDIR}/plugins \
-		--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots || exit 1; \
+		--repositories=https://central.sonatype.com/repository/maven-snapshots || exit 1; \
 		fi \
     done < $$PLUGIN_LIST
 
@@ -130,9 +130,6 @@ datasources:
     username: kestra
     password: k3str4
 kestra:
-  server:
-    basic-auth:
-    enabled: false
   encryption:
     secret-key: 3ywuDa/Ec61VHkOX3RlI9gYq7CaD0mv0Pf3DHtAXA6U=
   repository:

build-and-start-e2e-tests.sh

@@ -0,0 +1,46 @@
+#!/bin/bash
+set -e
+
+# E2E main script that can be run on a dev computer or in the CI
+# it will build the backend of the current git repo and the frontend
+# create a docker image out of it
+# run tests on this image
+
+
+LOCAL_IMAGE_VERSION="local-e2e"
+
+echo "Running E2E"
+echo "Start time: $(date '+%Y-%m-%d %H:%M:%S')"
+start_time=$(date +%s)
+
+echo ""
+echo "Building the image for this current repository"
+make build-docker VERSION=$LOCAL_IMAGE_VERSION
+
+end_time=$(date +%s)
+elapsed=$(( end_time - start_time ))
+
+echo ""
+echo "building elapsed time: ${elapsed} seconds"
+echo ""
+echo "Start time: $(date '+%Y-%m-%d %H:%M:%S')"
+start_time2=$(date +%s)
+
+echo "cd ./ui"
+cd ./ui
+echo "npm i"
+npm i
+
+echo 'sh ./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"'
+sh ./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"
+
+end_time2=$(date +%s)
+elapsed2=$(( end_time2 - start_time2 ))
+echo ""
+echo "Tests elapsed time: ${elapsed2} seconds"
+echo ""
+total_elapsed=$(( elapsed + elapsed2 ))
+echo "Total elapsed time: ${total_elapsed} seconds"
+echo ""
+
+exit 0

build.gradle

@@ -16,7 +16,7 @@ plugins {
     id "java"
     id 'java-library'
     id "idea"
-    id "com.gradleup.shadow" version "8.3.6"
+    id "com.gradleup.shadow" version "8.3.8"
     id "application"
 
     // test
@@ -31,12 +31,10 @@ plugins {
     id 'com.github.node-gradle.node' version '7.1.0'
 
     // release
-    id "io.github.gradle-nexus.publish-plugin" version "2.0.0"
     id 'net.researchgate.release' version '3.1.0'
-    id "com.gorylenko.gradle-git-properties" version "2.5.0"
+    id "com.gorylenko.gradle-git-properties" version "2.5.2"
     id 'signing'
-    id 'ru.vyarus.pom' version '3.0.0' apply false
-    id 'ru.vyarus.github-info' version '2.0.0' apply false
+    id "com.vanniktech.maven.publish" version "0.34.0"
 
     // OWASP dependency check
     id "org.owasp.dependencycheck" version "12.1.3" apply false
@@ -73,6 +71,11 @@ dependencies {
  * Dependencies
  **********************************************************************************************************************/
 allprojects {
+
+    tasks.withType(GenerateModuleMetadata).configureEach {
+        suppressedValidationErrors.add('enforced-platform')
+    }
+
     if (it.name != 'platform') {
         group = "io.kestra"
 
@@ -145,6 +148,7 @@ allprojects {
             implementation group: 'com.fasterxml.jackson.module', name: 'jackson-module-parameter-names'
             implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-guava'
             implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310'
+            implementation group: 'com.fasterxml.uuid', name: 'java-uuid-generator'
 
             // kestra
             implementation group: 'com.devskiller.friendly-id', name: 'friendly-id'
@@ -414,6 +418,7 @@ distTar.dependsOn shadowJar
 startScripts.dependsOn shadowJar
 startShadowScripts.dependsOn jar
 shadowJar.dependsOn 'ui:assembleFrontend'
+shadowJar.dependsOn jar
 
 /**********************************************************************************************************************\
  * Executable Jar
@@ -484,24 +489,11 @@ tasks.register('runStandalone', JavaExec) {
 /**********************************************************************************************************************\
  * Publish
  **********************************************************************************************************************/
-nexusPublishing {
-    repositoryDescription = "${project.group}:${rootProject.name}:${project.version}"
-    useStaging = !project.version.endsWith("-SNAPSHOT")
-    repositories {
-        sonatype {
-            nexusUrl.set(uri("https://s01.oss.sonatype.org/service/local/"))
-            snapshotRepositoryUrl.set(uri("https://s01.oss.sonatype.org/content/repositories/snapshots/"))
-        }
-    }
-}
+subprojects {subProject ->
 
-subprojects {
-
-    if (it.name != 'jmh-benchmarks') {
-        apply plugin: "maven-publish"
+    if (subProject.name != 'jmh-benchmarks' && subProject.name != rootProject.name) {
         apply plugin: 'signing'
-        apply plugin: 'ru.vyarus.pom'
-        apply plugin: 'ru.vyarus.github-info'
+        apply plugin: "com.vanniktech.maven.publish"
 
         javadoc {
             options {
@@ -535,66 +527,104 @@ subprojects {
             }
         }
 
-        github {
-            user 'kestra-io'
-            license 'Apache'
-            repository 'kestra'
-            site 'https://kestra.io'
+        //These modules should not be published
+        def unpublishedModules = ["jdbc-mysql", "jdbc-postgres", "webserver"]
+        if (subProject.name in unpublishedModules){
+            return
         }
 
-        maven.pom {
-            description = 'The modern, scalable orchestrator & scheduler open source platform'
+        mavenPublishing {
+            publishToMavenCentral(true)
+            signAllPublications()
 
-            developers {
-                developer {
-                    id = "tchiotludo"
-                    name = "Ludovic Dehon"
+            coordinates(
+                "${rootProject.group}",
+                subProject.name == "cli" ? rootProject.name : subProject.name,
+                "${rootProject.version}"
+            )
+
+            pom {
+                name        = project.name
+                description = "${project.group}:${project.name}:${rootProject.version}"
+                url         = "https://github.com/kestra-io/${rootProject.name}"
+
+                licenses {
+                    license {
+                        name = "The Apache License, Version 2.0"
+                        url  = "http://www.apache.org/licenses/LICENSE-2.0.txt"
+                    }
+                }
+                developers {
+                    developer {
+                        id    = "tchiotludo"
+                        name  = "Ludovic Dehon"
+                        email = "ldehon@kestra.io"
+                    }
+                }
+                scm {
+                    connection = 'scm:git:'
+                    url = "https://github.com/kestra-io/${rootProject.name}"
                 }
             }
         }
 
-        publishing {
-            publications {
-                sonatypePublication(MavenPublication) {
-                    version project.version
+        afterEvaluate {
+            publishing {
+                publications {
+                    withType(MavenPublication).configureEach { publication ->
 
-                    if (project.name.contains('cli')) {
-                        groupId "io.kestra"
-                        artifactId "kestra"
-
-                        artifact shadowJar
-                        artifact executableJar
-                    } else if (project.name.contains('platform')){
-                        groupId project.group
-                        artifactId project.name
-                    } else {
-                        from components.java
-
-                        groupId project.group
-                        artifactId project.name
-
-                        artifact sourcesJar
-                        artifact javadocJar
-                        artifact testsJar
+                        if (subProject.name == "platform") {
+                            // Clear all artifacts except the BOM
+                            publication.artifacts.clear()
+                        }
                     }
                 }
             }
         }
 
-        signing {
-            // only sign JARs that we publish to Sonatype
-            required { gradle.taskGraph.hasTask("publishSonatypePublicationPublicationToSonatypeRepository") }
-            sign publishing.publications.sonatypePublication
-        }
+        if (subProject.name == 'cli') {
 
-        tasks.withType(GenerateModuleMetadata).configureEach {
-            // Suppression this validation error as we want to enforce the Kestra platform
-            suppressedValidationErrors.add('enforced-platform')
+            /* Make sure the special publication is wired *after* every plugin */
+            subProject.afterEvaluate {
+                /* 1. Remove the default java component so Gradle stops expecting
+                       the standard cli-*.jar, sources, javadoc, etc.           */
+                components.removeAll { it.name == "java" }
+
+                /* 2. Replace the publication’s artifacts with shadow + exec */
+                publishing.publications.withType(MavenPublication).configureEach { pub ->
+                    pub.artifacts.clear()
+
+                    // main shadow JAR built at root
+                    pub.artifact(rootProject.tasks.named("shadowJar").get()) {
+                        extension = "jar"
+                    }
+
+                    // executable ZIP built at root
+                    pub.artifact(rootProject.tasks.named("executableJar").get().archiveFile) {
+                        classifier = "exec"
+                        extension  = "zip"
+                    }
+                    pub.artifact(tasks.named("sourcesJar").get())
+                    pub.artifact(tasks.named("javadocJar").get())
+
+                }
+
+                /* 3. Disable Gradle-module metadata for this publication to
+                       avoid the “artifact removed from java component” error. */
+                tasks.withType(GenerateModuleMetadata).configureEach { it.enabled = false }
+
+                /* 4. Make every publish task in :cli wait for the two artifacts */
+                tasks.matching { it.name.startsWith("publish") }.configureEach {
+                    dependsOn rootProject.tasks.named("shadowJar")
+                    dependsOn rootProject.tasks.named("executableJar")
+                }
+            }
         }
     }
 }
 
 
+
 /**********************************************************************************************************************\
  * Version
  **********************************************************************************************************************/

cli/build.gradle

@@ -37,4 +37,4 @@ dependencies {
 
     //test
     testImplementation "org.wiremock:wiremock-jetty12"
-}
+}

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

@@ -1,7 +1,5 @@
 package io.kestra.cli;
 
-import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
-
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.http.HttpHeaders;
 import io.micronaut.http.HttpRequest;
@@ -16,16 +14,15 @@ import io.micronaut.http.netty.body.NettyJsonHandler;
 import io.micronaut.json.JsonMapper;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
-import lombok.Builder;
-import lombok.Value;
-import lombok.extern.jackson.Jacksonized;
-import picocli.CommandLine;
-
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import lombok.Builder;
+import lombok.Value;
+import lombok.extern.jackson.Jacksonized;
+import picocli.CommandLine;
 
 public abstract class AbstractApiCommand extends AbstractCommand {
     @CommandLine.Option(names = {"--server"}, description = "Kestra server url", defaultValue = "http://localhost:8080")
@@ -37,7 +34,7 @@ public abstract class AbstractApiCommand extends AbstractCommand {
     @CommandLine.Option(names = {"--user"}, paramLabel = "<user:password>", description = "Server user and password")
     protected String user;
 
-    @CommandLine.Option(names = {"--tenant"}, description = "Tenant identifier (EE only, when multi-tenancy is enabled)")
+    @CommandLine.Option(names = {"--tenant"}, description = "Tenant identifier (EE only)")
     protected String tenantId;
 
     @CommandLine.Option(names = {"--api-token"}, description = "API Token (EE only).")
@@ -87,12 +84,12 @@ public abstract class AbstractApiCommand extends AbstractCommand {
         return request;
     }
 
-    protected String apiUri(String path) {
+    protected String apiUri(String path, String tenantId) {
         if (path == null || !path.startsWith("/")) {
             throw new IllegalArgumentException("'path' must be non-null and start with '/'");
         }
 
-        return tenantId == null ? "/api/v1/" + MAIN_TENANT + path : "/api/v1/" + tenantId + path;
+        return "/api/v1/" + tenantId + path;
     }
 
     @Builder

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

@@ -40,7 +40,7 @@ import picocli.CommandLine.Option;
 )
 @Slf4j
 @Introspected
-abstract public class AbstractCommand implements Callable<Integer> {
+public abstract class AbstractCommand implements Callable<Integer> {
     @Inject
     private ApplicationContext applicationContext;
 
@@ -93,7 +93,7 @@ abstract public class AbstractCommand implements Callable<Integer> {
             this.startupHook.start(this);
         }
 
-        if (this.pluginsPath != null && loadExternalPlugins()) {
+        if (pluginRegistryProvider != null && this.pluginsPath != null && loadExternalPlugins()) {
             pluginRegistry = pluginRegistryProvider.get();
             pluginRegistry.registerIfAbsent(pluginsPath);
 

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

@@ -1,5 +1,6 @@
 package io.kestra.cli;
 
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.models.validations.ModelValidator;
 import io.kestra.core.models.validations.ValidateConstraintViolation;
 import io.kestra.core.serializers.YamlParser;
@@ -9,6 +10,7 @@ import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import picocli.CommandLine;
 
 import java.io.IOException;
@@ -31,6 +33,9 @@ public abstract class AbstractValidateCommand extends AbstractApiCommand {
     @CommandLine.Parameters(index = "0", description = "the directory containing files to check")
     protected Path directory;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     /** {@inheritDoc} **/
     @Override
     protected boolean loadExternalPlugins() {
@@ -112,7 +117,7 @@ public abstract class AbstractValidateCommand extends AbstractApiCommand {
 
             try(DefaultHttpClient client = client()) {
                 MutableHttpRequest<String> request = HttpRequest
-                    .POST(apiUri("/flows/validate"), body).contentType(MediaType.APPLICATION_YAML);
+                    .POST(apiUri("/flows/validate", tenantService.getTenantId(tenantId)), body).contentType(MediaType.APPLICATION_YAML);
 
                 List<ValidateConstraintViolation> validations = client.toBlocking().retrieve(
                     this.requestOptions(request),

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

@@ -66,8 +66,14 @@ public class App implements Callable<Integer> {
         ApplicationContext applicationContext = App.applicationContext(cls, args);
 
         // Call Picocli command
-        int exitCode = new CommandLine(cls, new MicronautFactory(applicationContext)).execute(args);
-
+        int exitCode = 0;
+        try {
+             exitCode = new CommandLine(cls, new MicronautFactory(applicationContext)).execute(args);
+        } catch (CommandLine.InitializationException e){
+            System.err.println("Could not initialize picoli ComandLine, err: " + e.getMessage());
+            e.printStackTrace();
+            exitCode = 1;
+        }
         applicationContext.close();
 
         // exit code

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

@@ -2,11 +2,13 @@ package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
@@ -23,6 +25,9 @@ public class FlowCreateCommand extends AbstractApiCommand {
     @CommandLine.Parameters(index = "0", description = "The file containing the flow")
     public Path flowFile;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @SuppressWarnings("deprecation")
     @Override
     public Integer call() throws Exception {
@@ -34,7 +39,7 @@ public class FlowCreateCommand extends AbstractApiCommand {
 
         try(DefaultHttpClient client = client()) {
             MutableHttpRequest<String> request = HttpRequest
-                .POST(apiUri("/flows"), body).contentType(MediaType.APPLICATION_YAML);
+                .POST(apiUri("/flows", tenantService.getTenantId(tenantId)), body).contentType(MediaType.APPLICATION_YAML);
 
             client.toBlocking().retrieve(
                 this.requestOptions(request),

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

@@ -2,10 +2,12 @@ package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
@@ -23,6 +25,9 @@ public class FlowDeleteCommand extends AbstractApiCommand {
     @CommandLine.Parameters(index = "1", description = "The ID of the flow")
     public String id;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @SuppressWarnings("deprecation")
     @Override
     public Integer call() throws Exception {
@@ -30,7 +35,7 @@ public class FlowDeleteCommand extends AbstractApiCommand {
 
         try(DefaultHttpClient client = client()) {
             MutableHttpRequest<String> request = HttpRequest
-                .DELETE(apiUri("/flows/" + namespace + "/" + id ));
+                .DELETE(apiUri("/flows/" + namespace + "/" + id, tenantService.getTenantId(tenantId)));
 
             client.toBlocking().exchange(
                 this.requestOptions(request)

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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
-import io.micronaut.context.ApplicationContext;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.HttpResponse;
 import io.micronaut.http.MediaType;
@@ -25,9 +25,8 @@ import java.nio.file.Path;
 public class FlowExportCommand extends AbstractApiCommand {
     private static final String DEFAULT_FILE_NAME = "flows.zip";
 
-    // @FIXME: Keep it for bug in micronaut that need to have inject on top level command to inject on abstract classe
     @Inject
-    private ApplicationContext applicationContext;
+    private TenantIdSelectorService tenantService;
 
     @CommandLine.Option(names = {"--namespace"}, description = "The namespace of flows to export")
     public String namespace;
@@ -41,7 +40,7 @@ public class FlowExportCommand extends AbstractApiCommand {
 
         try(DefaultHttpClient client = client()) {
             MutableHttpRequest<Object> request = HttpRequest
-                .GET(apiUri("/flows/export/by-query") + (namespace != null ? "?namespace=" + namespace : ""))
+                .GET(apiUri("/flows/export/by-query", tenantService.getTenantId(tenantId)) + (namespace != null ? "?namespace=" + namespace : ""))
                 .accept(MediaType.APPLICATION_OCTET_STREAM);
 
             HttpResponse<byte[]> response = client.toBlocking().exchange(this.requestOptions(request), byte[].class);

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

@@ -1,7 +1,8 @@
 package io.kestra.cli.commands.flows;
 
 import com.google.common.collect.ImmutableMap;
-import io.kestra.cli.AbstractCommand;
+import io.kestra.cli.AbstractApiCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
@@ -30,7 +31,7 @@ import java.util.concurrent.TimeoutException;
     description = "Test a flow"
 )
 @Slf4j
-public class FlowTestCommand extends AbstractCommand {
+public class FlowTestCommand extends AbstractApiCommand {
     @Inject
     private ApplicationContext applicationContext;
 
@@ -76,6 +77,7 @@ public class FlowTestCommand extends AbstractCommand {
         FlowRepositoryInterface flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
         FlowInputOutput flowInputOutput = applicationContext.getBean(FlowInputOutput.class);
         RunnerUtils runnerUtils = applicationContext.getBean(RunnerUtils.class);
+        TenantIdSelectorService tenantService =  applicationContext.getBean(TenantIdSelectorService.class);
 
         Map<String, Object> inputs = new HashMap<>();
 
@@ -89,7 +91,7 @@ public class FlowTestCommand extends AbstractCommand {
 
         try {
             runner.run();
-            repositoryLoader.load(file.toFile());
+            repositoryLoader.load(tenantService.getTenantId(tenantId), file.toFile());
 
             List<Flow> all = flowRepository.findAllForAllTenants();
             if (all.size() != 1) {

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

@@ -2,11 +2,13 @@ package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
@@ -29,6 +31,9 @@ public class FlowUpdateCommand extends AbstractApiCommand {
     @CommandLine.Parameters(index = "2", description = "The ID of the flow")
     public String id;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @SuppressWarnings("deprecation")
     @Override
     public Integer call() throws Exception {
@@ -40,7 +45,7 @@ public class FlowUpdateCommand extends AbstractApiCommand {
 
         try(DefaultHttpClient client = client()) {
             MutableHttpRequest<String> request = HttpRequest
-                .PUT(apiUri("/flows/" + namespace + "/" + id ), body).contentType(MediaType.APPLICATION_YAML);
+                .PUT(apiUri("/flows/" + namespace + "/" + id, tenantService.getTenantId(tenantId)), body).contentType(MediaType.APPLICATION_YAML);
 
             client.toBlocking().retrieve(
                 this.requestOptions(request),

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

@@ -2,6 +2,7 @@ package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.serializers.YamlParser;
 import io.micronaut.core.type.Argument;
 import io.micronaut.http.HttpRequest;
@@ -9,6 +10,7 @@ import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import jakarta.validation.ConstraintViolationException;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
@@ -36,6 +38,9 @@ public class FlowUpdatesCommand extends AbstractApiCommand {
     @CommandLine.Option(names = {"--namespace"}, description = "The parent namespace of the flows, if not set, every namespace are allowed.")
     public String namespace;
 
+    @Inject
+    private TenantIdSelectorService tenantIdSelectorService;
+
     @SuppressWarnings("deprecation")
     @Override
     public Integer call() throws Exception {
@@ -66,7 +71,7 @@ public class FlowUpdatesCommand extends AbstractApiCommand {
                     namespaceQuery = "&namespace=" + namespace;
                 }
                 MutableHttpRequest<String> request = HttpRequest
-                    .POST(apiUri("/flows/bulk") + "?allowNamespaceChild=true&delete=" + delete + namespaceQuery, body).contentType(MediaType.APPLICATION_YAML);
+                    .POST(apiUri("/flows/bulk", tenantIdSelectorService.getTenantId(tenantId)) + "?allowNamespaceChild=true&delete=" + delete + namespaceQuery, body).contentType(MediaType.APPLICATION_YAML);
 
                 List<UpdateResult> updated = client.toBlocking().retrieve(
                     this.requestOptions(request),

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

@@ -1,6 +1,7 @@
 package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.validations.ModelValidator;
 import io.kestra.core.services.FlowService;
@@ -22,6 +23,9 @@ public class FlowValidateCommand extends AbstractValidateCommand {
     @Inject
     private FlowService flowService;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @Override
     public Integer call() throws Exception {
         return this.call(
@@ -35,7 +39,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, this.tenantId));
+                warnings.addAll(flowService.warnings(flow, tenantService.getTenantId(tenantId)));
                 return warnings;
             },
             (Object object) -> {

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

@@ -3,6 +3,7 @@ package io.kestra.cli.commands.flows.namespaces;
 import io.kestra.cli.AbstractValidateCommand;
 import io.kestra.cli.commands.AbstractServiceNamespaceUpdateCommand;
 import io.kestra.cli.commands.flows.IncludeHelperExpander;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.serializers.YamlParser;
 import io.micronaut.core.type.Argument;
 import io.micronaut.http.HttpRequest;
@@ -10,6 +11,7 @@ import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import jakarta.validation.ConstraintViolationException;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
@@ -30,6 +32,9 @@ public class FlowNamespaceUpdateCommand extends AbstractServiceNamespaceUpdateCo
     @CommandLine.Option(names = {"--override-namespaces"}, negatable = true, description = "Replace namespace of all flows by the one provided")
     public boolean override = false;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @SuppressWarnings("deprecation")
     @Override
     public Integer call() throws Exception {
@@ -59,7 +64,7 @@ public class FlowNamespaceUpdateCommand extends AbstractServiceNamespaceUpdateCo
             }
             try(DefaultHttpClient client = client()) {
                 MutableHttpRequest<String> request = HttpRequest
-                    .POST(apiUri("/flows/") + namespace + "?delete=" + delete, body).contentType(MediaType.APPLICATION_YAML);
+                    .POST(apiUri("/flows/", tenantService.getTenantId(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/namespaces/files/NamespaceFilesUpdateCommand.java

@@ -2,12 +2,14 @@ package io.kestra.cli.commands.namespaces.files;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.utils.KestraIgnore;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.multipart.MultipartBody;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
@@ -34,6 +36,9 @@ public class NamespaceFilesUpdateCommand extends AbstractApiCommand {
     @CommandLine.Option(names = {"--delete"}, negatable = true, description = "Whether missing should be deleted")
     public boolean delete = false;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     private static final String KESTRA_IGNORE_FILE = ".kestraignore";
 
     @Override
@@ -44,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/") + namespace + "/files?path=" + to, null)));
+                client.toBlocking().exchange(this.requestOptions(HttpRequest.DELETE(apiUri("/namespaces/", tenantService.getTenantId(tenantId)) + namespace + "/files?path=" + to, null)));
             }
 
             KestraIgnore kestraIgnore = new KestraIgnore(from);
@@ -62,7 +67,7 @@ public class NamespaceFilesUpdateCommand extends AbstractApiCommand {
                 client.toBlocking().exchange(
                     this.requestOptions(
                         HttpRequest.POST(
-                            apiUri("/namespaces/") + namespace + "/files?path=" + destination,
+                            apiUri("/namespaces/", tenantService.getTenantId(tenantId)) + namespace + "/files?path=" + destination,
                             body
                         ).contentType(MediaType.MULTIPART_FORM_DATA)
                     )

cli/src/main/java/io/kestra/cli/commands/namespaces/kv/KvUpdateCommand.java

@@ -3,11 +3,13 @@ package io.kestra.cli.commands.namespaces.kv;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.kestra.cli.AbstractApiCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.serializers.JacksonMapper;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 import picocli.CommandLine.Option;
@@ -42,6 +44,9 @@ public class KvUpdateCommand extends AbstractApiCommand {
     @Option(names = {"-f", "--file-value"}, description = "The file from which to read the value to set. If this is provided, it will take precedence over any specified value.")
     public Path fileValue;
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @Override
     public Integer call() throws Exception {
         super.call();
@@ -56,7 +61,7 @@ public class KvUpdateCommand extends AbstractApiCommand {
 
         Duration ttl = expiration == null ? null : Duration.parse(expiration);
         MutableHttpRequest<String> request = HttpRequest
-            .PUT(apiUri("/namespaces/") + namespace + "/kv/" + key, value)
+            .PUT(apiUri("/namespaces/", tenantService.getTenantId(tenantId)) + namespace + "/kv/" + key, value)
             .contentType(MediaType.APPLICATION_JSON_TYPE);
 
         if (ttl != null) {

cli/src/main/java/io/kestra/cli/commands/plugins/PluginDocCommand.java

@@ -18,6 +18,8 @@ import java.nio.file.Paths;
 import java.util.Base64;
 import java.util.List;
 
+import static io.kestra.core.models.Plugin.isDeprecated;
+
 @CommandLine.Command(
     name = "doc",
     description = "Generate documentation for all plugins currently installed"
@@ -38,6 +40,9 @@ public class PluginDocCommand extends AbstractCommand {
     @CommandLine.Option(names = {"--schema"}, description = "Also write JSON Schema for each task")
     private boolean schema = false;
 
+    @CommandLine.Option(names = {"--skip-deprecated"},description = "Skip deprecated plugins when generating documentations")
+    private boolean skipDeprecated = false;
+
     @Override
     public Integer call() throws Exception {
         super.call();
@@ -45,6 +50,11 @@ public class PluginDocCommand extends AbstractCommand {
 
         PluginRegistry registry = pluginRegistryProvider.get();
         List<RegisteredPlugin> plugins = core ? registry.plugins() : registry.externalPlugins();
+        if (skipDeprecated) {
+            plugins = plugins.stream()
+                .filter(plugin -> !isDeprecated(plugin.getClass()))
+                .toList();
+        }
         boolean hasFailures = false;
 
         for (RegisteredPlugin registeredPlugin : plugins) {

cli/src/main/java/io/kestra/cli/commands/servers/StandAloneCommand.java

@@ -2,6 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.cli.services.FileChangedEventListener;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
@@ -44,6 +45,9 @@ public class StandAloneCommand extends AbstractServerCommand {
     @CommandLine.Option(names = {"-f", "--flow-path"}, description = "the flow path containing flow to inject at startup (when running with a memory flow repository)")
     private File flowPath;
 
+    @CommandLine.Option(names = "--tenant", description = "Tenant identifier, Required to load flows from path with the enterprise edition")
+    private String tenantId;
+
     @CommandLine.Option(names = {"--worker-thread"}, description = "the number of worker threads, defaults to four times the number of available processors. Set it to 0 to avoid starting a worker.")
     private int workerThread = defaultWorkerThread();
 
@@ -98,7 +102,8 @@ public class StandAloneCommand extends AbstractServerCommand {
         if (flowPath != null) {
             try {
                 LocalFlowRepositoryLoader localFlowRepositoryLoader = applicationContext.getBean(LocalFlowRepositoryLoader.class);
-                localFlowRepositoryLoader.load(null, this.flowPath);
+                TenantIdSelectorService tenantIdSelectorService = applicationContext.getBean(TenantIdSelectorService.class);
+                localFlowRepositoryLoader.load(tenantIdSelectorService.getTenantId(this.tenantId), this.flowPath);
             } catch (IOException e) {
                 throw new CommandLine.ParameterException(this.spec.commandLine(), "Invalid flow path", e);
             }

cli/src/main/java/io/kestra/cli/commands/templates/TemplateExportCommand.java

@@ -2,8 +2,8 @@ package io.kestra.cli.commands.templates;
 
 import io.kestra.cli.AbstractApiCommand;
 import io.kestra.cli.AbstractValidateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.models.templates.TemplateEnabled;
-import io.micronaut.context.ApplicationContext;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.HttpResponse;
 import io.micronaut.http.MediaType;
@@ -27,9 +27,8 @@ import java.nio.file.Path;
 public class TemplateExportCommand extends AbstractApiCommand {
     private static final String DEFAULT_FILE_NAME = "templates.zip";
 
-    // @FIXME: Keep it for bug in micronaut that need to have inject on top level command to inject on abstract classe
     @Inject
-    private ApplicationContext applicationContext;
+    private TenantIdSelectorService tenantService;
 
     @CommandLine.Option(names = {"--namespace"}, description = "The namespace of templates to export")
     public String namespace;
@@ -43,7 +42,7 @@ public class TemplateExportCommand extends AbstractApiCommand {
 
         try(DefaultHttpClient client = client()) {
             MutableHttpRequest<Object> request = HttpRequest
-                .GET(apiUri("/templates/export/by-query") + (namespace != null ? "?namespace=" + namespace : ""))
+                .GET(apiUri("/templates/export/by-query", tenantService.getTenantId(tenantId)) + (namespace != null ? "?namespace=" + namespace : ""))
                 .accept(MediaType.APPLICATION_OCTET_STREAM);
 
             HttpResponse<byte[]> response = client.toBlocking().exchange(this.requestOptions(request), byte[].class);

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

@@ -2,6 +2,7 @@ package io.kestra.cli.commands.templates.namespaces;
 
 import io.kestra.cli.AbstractValidateCommand;
 import io.kestra.cli.commands.AbstractServiceNamespaceUpdateCommand;
+import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.models.templates.Template;
 import io.kestra.core.models.templates.TemplateEnabled;
 import io.kestra.core.serializers.YamlParser;
@@ -10,6 +11,7 @@ import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
+import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
@@ -27,6 +29,9 @@ import jakarta.validation.ConstraintViolationException;
 @TemplateEnabled
 public class TemplateNamespaceUpdateCommand extends AbstractServiceNamespaceUpdateCommand {
 
+    @Inject
+    private TenantIdSelectorService tenantService;
+
     @Override
     public Integer call() throws Exception {
         super.call();
@@ -44,7 +49,7 @@ public class TemplateNamespaceUpdateCommand extends AbstractServiceNamespaceUpda
 
             try (DefaultHttpClient client = client()) {
                 MutableHttpRequest<List<Template>> request = HttpRequest
-                    .POST(apiUri("/templates/") + namespace + "?delete=" + delete, templates);
+                    .POST(apiUri("/templates/", tenantService.getTenantId(tenantId)) + namespace + "?delete=" + delete, templates);
 
                 List<UpdateResult> updated = client.toBlocking().retrieve(
                     this.requestOptions(request),

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

@@ -0,0 +1,19 @@
+package io.kestra.cli.services;
+
+import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
+
+import io.kestra.core.exceptions.KestraRuntimeException;
+import jakarta.inject.Singleton;
+import org.apache.commons.lang3.StringUtils;
+
+@Singleton
+public class TenantIdSelectorService {
+
+    //For override purpose in Kestra EE
+    public String getTenantId(String tenantId) {
+        if (StringUtils.isNotBlank(tenantId) && !MAIN_TENANT.equals(tenantId)){
+            throw new KestraRuntimeException("Tenant id can only be 'main'");
+        }
+        return MAIN_TENANT;
+    }
+}

cli/src/main/resources/application.yml

@@ -27,6 +27,7 @@ micronaut:
     write-idle-timeout: 60m
     idle-timeout: 60m
     netty:
+      max-zstd-encode-size: 67108864 # increased to 64MB from the default of 32MB
       max-chunk-size: 10MB
       max-header-size: 32768 # increased from the default of 8k
     responses:
@@ -183,7 +184,6 @@ kestra:
 
   server:
     basic-auth:
-      enabled: false
       # These URLs will not be authenticated, by default we open some of the Micronaut default endpoints but not all for security reasons
       open-urls:
         - "/ping"
@@ -228,4 +228,4 @@ otel:
     - /health
     - /env
     - /prometheus
-  propagators: tracecontext, baggage
+  propagators: tracecontext, baggage

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

@@ -108,6 +108,34 @@ class FlowCreateOrUpdateCommandTest {
         }
     }
 
+    @Test
+    void should_fail_with_incorrect_tenant()  {
+        URL directory = FlowCreateOrUpdateCommandTest.class.getClassLoader().getResource("flows");
+
+        ByteArrayOutputStream err = new ByteArrayOutputStream();
+        System.setErr(new PrintStream(err));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+
+            EmbeddedServer embeddedServer = ctx.getBean(EmbeddedServer.class);
+            embeddedServer.start();
+
+            String[] args = {
+                "--server",
+                embeddedServer.getURL().toString(),
+                "--user",
+                "myuser:pass:word",
+                "--tenant", "incorrect",
+                directory.getPath(),
+
+            };
+            PicocliRunner.call(FlowUpdatesCommand.class, ctx, args);
+
+            assertThat(err.toString()).contains("Tenant id can only be 'main'");
+            err.reset();
+        }
+    }
+
     @Test
     void helper()  {
         URL directory = FlowCreateOrUpdateCommandTest.class.getClassLoader().getResource("helper");

cli/src/test/java/io/kestra/cli/commands/servers/TenantIdSelectorServiceTest.java

@@ -0,0 +1,33 @@
+package io.kestra.cli.commands.servers;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.kestra.cli.App;
+import io.micronaut.configuration.picocli.PicocliRunner;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.env.Environment;
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import org.junit.jupiter.api.Test;
+
+public class TenantIdSelectorServiceTest {
+
+    @Test
+    void should_fail_without_tenant_id() {
+        ByteArrayOutputStream err = new ByteArrayOutputStream();
+        System.setErr(new PrintStream(err));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+            String[] start = {
+                "server", "standalone",
+                "-f", "unused",
+                "--tenant", "wrong_tenant"
+            };
+            PicocliRunner.call(App.class, ctx, start);
+
+            assertThat(err.toString()).contains("Tenant id can only be 'main'");
+            err.reset();
+        }
+    }
+
+}

cli/src/test/resources/application-test.yml

@@ -17,7 +17,7 @@ kestra:
       central:
         url: https://repo.maven.apache.org/maven2/
       sonatype:
-        url: https://s01.oss.sonatype.org/content/repositories/snapshots/
+        url: https://central.sonatype.com/repository/maven-snapshots/
   server:
     liveness:
       enabled: false

codecov.yml

@@ -56,21 +56,23 @@ component_management:
       name: Tests
       paths:
         - tests/**
-    - component_id: ui
-      name: Ui
-      paths:
-        - ui/**
+
     - component_id: webserver
       name: Webserver
       paths:
         - webserver/**
 
+ignore:
+  - ui/**
+# we are not mature yet to have a ui code coverage
+
 flag_management:
   default_rules:
     carryforward: true
     statuses:
       - type: project
-        target: 80%
-        threshold: 1%
+        target: 70%
+        threshold: 10%
       - type: patch
-        target: 90%
+        target: 75%
+        threshold: 10%

core/build.gradle

@@ -37,6 +37,7 @@ dependencies {
     implementation 'nl.basjes.gitignore:gitignore-reader'
     implementation group: 'dev.failsafe', name: 'failsafe'
     implementation 'com.github.ben-manes.caffeine:caffeine'
+    implementation 'com.github.ksuid:ksuid:1.1.3'
     api 'org.apache.httpcomponents.client5:httpclient5'
 
     // plugins
@@ -74,7 +75,9 @@ dependencies {
     testImplementation "io.micronaut:micronaut-http-server-netty"
     testImplementation "io.micronaut:micronaut-management"
 
-    testImplementation "org.testcontainers:testcontainers:1.21.1"
-    testImplementation "org.testcontainers:junit-jupiter:1.21.1"
+    testImplementation "org.testcontainers:testcontainers:1.21.3"
+    testImplementation "org.testcontainers:junit-jupiter:1.21.3"
     testImplementation "org.bouncycastle:bcpkix-jdk18on:1.81"
+
+    testImplementation "org.wiremock:wiremock-jetty12"
 }

core/src/main/java/io/kestra/core/debug/Breakpoint.java

@@ -0,0 +1,26 @@
+package io.kestra.core.debug;
+
+import jakarta.annotation.Nullable;
+import jakarta.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+@AllArgsConstructor
+@NoArgsConstructor
+@Getter
+public class Breakpoint {
+    @NotNull
+    private String id;
+
+    @Nullable
+    private String value;
+
+    public static Breakpoint of(String breakpoint) {
+        if (breakpoint.indexOf('.') > 0) {
+            return new Breakpoint(breakpoint.substring(0, breakpoint.indexOf('.')), breakpoint.substring(breakpoint.indexOf('.') + 1));
+        } else {
+            return new Breakpoint(breakpoint, null);
+        }
+    }
+}

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

@@ -6,14 +6,17 @@ import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.ToString;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 
 @Getter
 @EqualsAndHashCode
 @ToString
 public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
+    private static final Map<PluginDocIdentifier, ClassPluginDocumentation<?>> CACHE = new ConcurrentHashMap<>();
     private String icon;
     private String group;
     protected String docLicense;
@@ -78,8 +81,12 @@ public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
         }
     }
 
-    public static <T> ClassPluginDocumentation<T> of(JsonSchemaGenerator jsonSchemaGenerator, PluginClassAndMetadata<T> plugin, boolean allProperties) {
-        return new ClassPluginDocumentation<>(jsonSchemaGenerator, plugin, allProperties);
+    public static <T> ClassPluginDocumentation<T> of(JsonSchemaGenerator jsonSchemaGenerator, PluginClassAndMetadata<T> plugin, String version, boolean allProperties) {
+        //noinspection unchecked
+        return (ClassPluginDocumentation<T>) CACHE.computeIfAbsent(
+            new PluginDocIdentifier(plugin.type(), version, allProperties),
+            (key) -> new ClassPluginDocumentation<>(jsonSchemaGenerator, plugin, allProperties)
+        );
     }
 
     @AllArgsConstructor
@@ -90,5 +97,11 @@ public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
         String unit;
         String description;
     }
+
+    private record PluginDocIdentifier(String pluginClassAndVersion, boolean allProperties) {
+        public PluginDocIdentifier(Class<?> pluginClass, String version, boolean allProperties) {
+            this(pluginClass.getName() + ":" + version, allProperties);
+        }
+    }
 }
 

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

@@ -227,7 +227,7 @@ public class DocumentationGenerator {
                     baseCls,
                     null
                 );
-                return ClassPluginDocumentation.of(jsonSchemaGenerator, metadata, true);
+                return ClassPluginDocumentation.of(jsonSchemaGenerator, metadata, registeredPlugin.version(), true);
             })
             .map(pluginDocumentation -> {
                 try {

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

@@ -24,6 +24,7 @@ public class JsonSchemaCache {
     private final JsonSchemaGenerator jsonSchemaGenerator;
 
     private final ConcurrentMap<CacheKey, Map<String, Object>> schemaCache = new ConcurrentHashMap<>();
+    private final ConcurrentMap<SchemaType, Map<String, Object>> propertiesCache = new ConcurrentHashMap<>();
 
     private final Map<SchemaType, Class<?>> classesBySchemaType = new HashMap<>();
 
@@ -44,7 +45,7 @@ public class JsonSchemaCache {
 
     public Map<String, Object> getSchemaForType(final SchemaType type,
                                                 final boolean arrayOf) {
-        return schemaCache.computeIfAbsent(new CacheKey(type, arrayOf), (key) -> {
+        return schemaCache.computeIfAbsent(new CacheKey(type, arrayOf), key -> {
 
             Class<?> cls = Optional.ofNullable(classesBySchemaType.get(type))
                 .orElseThrow(() -> new IllegalArgumentException("Cannot found schema for type '" + type + "'"));
@@ -52,6 +53,16 @@ public class JsonSchemaCache {
         });
     }
 
+    public Map<String, Object> getPropertiesForType(final SchemaType type) {
+        return propertiesCache.computeIfAbsent(type, key -> {
+
+            Class<?> cls = Optional.ofNullable(classesBySchemaType.get(type))
+                .orElseThrow(() -> new IllegalArgumentException("Cannot found properties for type '" + type + "'"));
+            return jsonSchemaGenerator.properties(null, cls);
+        });
+    }
+
+    // must be public as it's used in EE
     public void registerClassForType(final SchemaType type, final Class<?> clazz) {
         classesBySchemaType.put(type, clazz);
     }

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

@@ -88,12 +88,16 @@ public class JsonSchemaGenerator {
     }
 
     public <T> Map<String, Object> schemas(Class<? extends T> cls, boolean arrayOf) {
+        return this.schemas(cls, arrayOf, Collections.emptyList());
+    }
+
+    public <T> Map<String, Object> schemas(Class<? extends T> cls, boolean arrayOf, List<String> allowedPluginTypes) {
         SchemaGeneratorConfigBuilder builder = new SchemaGeneratorConfigBuilder(
             SchemaVersion.DRAFT_7,
             OptionPreset.PLAIN_JSON
         );
 
-        this.build(builder, true);
+        this.build(builder, true, allowedPluginTypes);
 
         SchemaGeneratorConfig schemaGeneratorConfig = builder.build();
 
@@ -240,6 +244,10 @@ public class JsonSchemaGenerator {
     }
 
     protected void build(SchemaGeneratorConfigBuilder builder, boolean draft7) {
+        this.build(builder, draft7, Collections.emptyList());
+    }
+
+    protected void build(SchemaGeneratorConfigBuilder builder, boolean draft7, List<String> allowedPluginTypes) {
 //        builder.withObjectMapper(builder.getObjectMapper().configure(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS, false));
         builder
             .with(new JakartaValidationModule(
@@ -456,7 +464,7 @@ public class JsonSchemaGenerator {
                 .withSubtypeResolver((declaredType, context) -> {
                     TypeContext typeContext = context.getTypeContext();
 
-                    return this.subtypeResolver(declaredType, typeContext);
+                    return this.subtypeResolver(declaredType, typeContext, allowedPluginTypes);
                 });
 
             // description as Markdown
@@ -533,7 +541,7 @@ public class JsonSchemaGenerator {
                         return null;
                     }
 
-                    return this.subtypeResolver(declaredType, typeContext);
+                    return this.subtypeResolver(declaredType, typeContext, allowedPluginTypes);
                 });
         }
 
@@ -616,11 +624,12 @@ public class JsonSchemaGenerator {
         return false;
     }
 
-    protected List<ResolvedType> subtypeResolver(ResolvedType declaredType, TypeContext typeContext) {
+    protected List<ResolvedType> subtypeResolver(ResolvedType declaredType, TypeContext typeContext, List<String> allowedPluginTypes) {
         if (declaredType.getErasedType() == Task.class) {
             return getRegisteredPlugins()
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getTasks().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .flatMap(clz -> safelyResolveSubtype(declaredType, clz, typeContext).stream())
                 .toList();
@@ -628,6 +637,7 @@ public class JsonSchemaGenerator {
             return getRegisteredPlugins()
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getTriggers().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .flatMap(clz -> safelyResolveSubtype(declaredType, clz, typeContext).stream())
                 .toList();
@@ -635,6 +645,7 @@ public class JsonSchemaGenerator {
             return getRegisteredPlugins()
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getConditions().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .flatMap(clz -> safelyResolveSubtype(declaredType, clz, typeContext).stream())
                 .toList();
@@ -643,6 +654,7 @@ public class JsonSchemaGenerator {
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getConditions().stream())
                 .filter(ScheduleCondition.class::isAssignableFrom)
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .flatMap(clz -> safelyResolveSubtype(declaredType, clz, typeContext).stream())
                 .toList();
@@ -650,6 +662,7 @@ public class JsonSchemaGenerator {
             return getRegisteredPlugins()
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getTaskRunners().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .map(typeContext::resolve)
                 .toList();
@@ -657,6 +670,7 @@ public class JsonSchemaGenerator {
             return getRegisteredPlugins()
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getLogExporters().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .map(typeContext::resolve)
                 .toList();
@@ -666,6 +680,7 @@ public class JsonSchemaGenerator {
                 .flatMap(registeredPlugin -> registeredPlugin.getAdditionalPlugins().stream())
                 // for additional plugins, we have one subtype by type of additional plugins (for ex: embedding store for Langchain4J), so we need to filter on the correct subtype
                 .filter(cls -> declaredType.getErasedType().isAssignableFrom(cls))
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(cls -> cls != declaredType.getErasedType())
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .map(typeContext::resolve)
@@ -674,6 +689,7 @@ public class JsonSchemaGenerator {
             return getRegisteredPlugins()
                 .stream()
                 .flatMap(registeredPlugin -> registeredPlugin.getCharts().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .<ResolvedType>mapMulti((clz, consumer) -> {
                     if (DataChart.class.isAssignableFrom(clz)) {
@@ -740,12 +756,16 @@ public class JsonSchemaGenerator {
     }
 
     protected <T> Map<String, Object> generate(Class<? extends T> cls, @Nullable Class<T> base) {
+        return this.generate(cls, base, Collections.emptyList());
+    }
+
+    protected <T> Map<String, Object> generate(Class<? extends T> cls, @Nullable Class<T> base, List<String> allowedPluginTypes) {
         SchemaGeneratorConfigBuilder builder = new SchemaGeneratorConfigBuilder(
             SchemaVersion.DRAFT_2019_09,
             OptionPreset.PLAIN_JSON
         );
 
-        this.build(builder, false);
+        this.build(builder, false, allowedPluginTypes);
 
         // we don't return base properties unless specified with @PluginProperty and hidden is false
         builder

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

@@ -23,29 +23,25 @@ public class Plugin {
     private String group;
     private String version;
     private Map<String, String> manifest;
-    private List<String> tasks;
-    private List<String> triggers;
-    private List<String> conditions;
-    private List<String> controllers;
-    private List<String> storages;
-    private List<String> secrets;
-    private List<String> taskRunners;
     private List<String> guides;
     private List<String> aliases;
-    private List<String> apps;
-    private List<String> appBlocks;
-    private List<String> charts;
-    private List<String> dataFilters;
-    private List<String> logExporters;
-    private List<String> additionalPlugins;
+    private List<PluginElementMetadata> tasks;
+    private List<PluginElementMetadata> triggers;
+    private List<PluginElementMetadata> conditions;
+    private List<PluginElementMetadata> controllers;
+    private List<PluginElementMetadata> storages;
+    private List<PluginElementMetadata> secrets;
+    private List<PluginElementMetadata> taskRunners;
+    private List<PluginElementMetadata> apps;
+    private List<PluginElementMetadata> appBlocks;
+    private List<PluginElementMetadata> charts;
+    private List<PluginElementMetadata> dataFilters;
+    private List<PluginElementMetadata> logExporters;
+    private List<PluginElementMetadata> additionalPlugins;
     private List<PluginSubGroup.PluginCategory> categories;
     private String subGroup;
 
     public static Plugin of(RegisteredPlugin registeredPlugin, @Nullable String subgroup) {
-        return Plugin.of(registeredPlugin, subgroup, true);
-    }
-
-    public static Plugin of(RegisteredPlugin registeredPlugin, @Nullable String subgroup, boolean includeDeprecated) {
         Plugin plugin = new Plugin();
         plugin.name = registeredPlugin.name();
         PluginSubGroup subGroupInfos = null;
@@ -90,18 +86,18 @@ public class Plugin {
         plugin.subGroup = subgroup;
 
         Predicate<Class<?>> packagePredicate = c -> subgroup == null || c.getPackageName().equals(subgroup);
-        plugin.tasks = filterAndGetClassName(registeredPlugin.getTasks(), includeDeprecated, packagePredicate);
-        plugin.triggers = filterAndGetClassName(registeredPlugin.getTriggers(), includeDeprecated, packagePredicate);
-        plugin.conditions = filterAndGetClassName(registeredPlugin.getConditions(), includeDeprecated, packagePredicate);
-        plugin.storages = filterAndGetClassName(registeredPlugin.getStorages(), includeDeprecated, packagePredicate);
-        plugin.secrets = filterAndGetClassName(registeredPlugin.getSecrets(), includeDeprecated, packagePredicate);
-        plugin.taskRunners = filterAndGetClassName(registeredPlugin.getTaskRunners(), includeDeprecated, packagePredicate);
-        plugin.apps = filterAndGetClassName(registeredPlugin.getApps(), includeDeprecated, packagePredicate);
-        plugin.appBlocks = filterAndGetClassName(registeredPlugin.getAppBlocks(), includeDeprecated, packagePredicate);
-        plugin.charts = filterAndGetClassName(registeredPlugin.getCharts(), includeDeprecated, packagePredicate);
-        plugin.dataFilters = filterAndGetClassName(registeredPlugin.getDataFilters(), includeDeprecated, packagePredicate);
-        plugin.logExporters = filterAndGetClassName(registeredPlugin.getLogExporters(), includeDeprecated, packagePredicate);
-        plugin.additionalPlugins = filterAndGetClassName(registeredPlugin.getAdditionalPlugins(), includeDeprecated, packagePredicate);
+        plugin.tasks = filterAndGetTypeWithMetadata(registeredPlugin.getTasks(), packagePredicate);
+        plugin.triggers = filterAndGetTypeWithMetadata(registeredPlugin.getTriggers(), packagePredicate);
+        plugin.conditions = filterAndGetTypeWithMetadata(registeredPlugin.getConditions(), packagePredicate);
+        plugin.storages = filterAndGetTypeWithMetadata(registeredPlugin.getStorages(), packagePredicate);
+        plugin.secrets = filterAndGetTypeWithMetadata(registeredPlugin.getSecrets(), packagePredicate);
+        plugin.taskRunners = filterAndGetTypeWithMetadata(registeredPlugin.getTaskRunners(), packagePredicate);
+        plugin.apps = filterAndGetTypeWithMetadata(registeredPlugin.getApps(), packagePredicate);
+        plugin.appBlocks = filterAndGetTypeWithMetadata(registeredPlugin.getAppBlocks(), packagePredicate);
+        plugin.charts = filterAndGetTypeWithMetadata(registeredPlugin.getCharts(), packagePredicate);
+        plugin.dataFilters = filterAndGetTypeWithMetadata(registeredPlugin.getDataFilters(), packagePredicate);
+        plugin.logExporters = filterAndGetTypeWithMetadata(registeredPlugin.getLogExporters(), packagePredicate);
+        plugin.additionalPlugins = filterAndGetTypeWithMetadata(registeredPlugin.getAdditionalPlugins(), packagePredicate);
 
         return plugin;
     }
@@ -111,17 +107,18 @@ public class Plugin {
      * Those classes are only filtered from the documentation to ensure backward compatibility.
      *
      * @param list              The list of classes?
-     * @param includeDeprecated whether to include deprecated plugins or not
      * @return a filtered streams.
      */
-    private static List<String> filterAndGetClassName(final List<? extends Class<?>> list, boolean includeDeprecated, Predicate<Class<?>> clazzFilter) {
+    private static List<PluginElementMetadata> filterAndGetTypeWithMetadata(final List<? extends Class<?>> list, Predicate<Class<?>> clazzFilter) {
         return list
             .stream()
             .filter(not(io.kestra.core.models.Plugin::isInternal))
-            .filter(p -> includeDeprecated || !io.kestra.core.models.Plugin.isDeprecated(p))
             .filter(clazzFilter)
-            .map(Class::getName)
-            .filter(c -> !c.startsWith("org.kestra."))
+            .filter(c -> !c.getName().startsWith("org.kestra."))
+            .map(c -> new PluginElementMetadata(c.getName(), io.kestra.core.models.Plugin.isDeprecated(c) ? true : null))
             .toList();
     }
+
+    public record PluginElementMetadata(String cls, Boolean deprecated) {
+    }
 }

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

@@ -0,0 +1,25 @@
+package io.kestra.core.exceptions;
+
+/**
+ * General exception that can be thrown when an AI service replies with an error.
+ * When propagated in the context of a REST API call, this exception should
+ * result in an HTTP 422 UNPROCESSABLE_ENTITY response.
+ */
+public class AiException extends KestraRuntimeException {
+
+    /**
+     * Creates a new {@link AiException} instance.
+     */
+    public AiException() {
+        super();
+    }
+
+    /**
+     * Creates a new {@link AiException} instance.
+     *
+     * @param aiErrorMessage the AI error message.
+     */
+    public AiException(final String aiErrorMessage) {
+        super(aiErrorMessage);
+    }
+}

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

@@ -0,0 +1,35 @@
+package io.kestra.core.exceptions;
+
+import java.io.Serial;
+import java.util.List;
+import lombok.Getter;
+
+/**
+ * General exception that can be throws when a resource fail validation.
+ */
+public class ValidationErrorException extends KestraRuntimeException {
+    @Serial
+    private static final long serialVersionUID = 1L;
+    private static final String VALIDATION_ERROR_MESSAGE = "Resource fails validation";
+
+    @Getter
+    private transient final List<String> invalids;
+
+    /**
+     * Creates a new {@link ValidationErrorException} instance.
+     *
+     * @param invalids the invalid filters.
+     */
+    public ValidationErrorException(final List<String> invalids) {
+        super(VALIDATION_ERROR_MESSAGE);
+        this.invalids = invalids;
+    }
+
+
+    public String formatedInvalidObjects(){
+        if (invalids == null || invalids.isEmpty()){
+            return VALIDATION_ERROR_MESSAGE;
+        }
+        return String.join(", ", invalids);
+    }
+}

core/src/main/java/io/kestra/core/http/client/HttpClient.java

@@ -8,6 +8,7 @@ import io.kestra.core.http.client.apache.*;
 import io.kestra.core.http.client.configurations.HttpConfiguration;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
+import io.micronaut.http.MediaType;
 import jakarta.annotation.Nullable;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
@@ -279,10 +280,12 @@ public class HttpClient implements Closeable {
     private <T> T bodyHandler(Class<?> cls, HttpEntity entity) throws IOException, ParseException {
         if (entity == null) {
             return null;
-        } else if (cls.isAssignableFrom(String.class)) {
+        } else if (String.class.isAssignableFrom(cls)) {
             return (T) EntityUtils.toString(entity);
-        } else if (cls.isAssignableFrom(Byte[].class)) {
+        } else if (Byte[].class.isAssignableFrom(cls)) {
             return (T) ArrayUtils.toObject(EntityUtils.toByteArray(entity));
+        } else if (MediaType.APPLICATION_YAML.equals(entity.getContentType()) || "application/yaml".equals(entity.getContentType())) {
+            return (T) JacksonMapper.ofYaml().readValue(entity.getContent(), cls);
         } else {
             return (T) JacksonMapper.ofJson(false).readValue(entity.getContent(), cls);
         }

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

@@ -9,6 +9,7 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Streams;
+import io.kestra.core.debug.Breakpoint;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.Label;
@@ -120,6 +121,9 @@ public class Execution implements DeletedInterface, TenantInterface {
     @Nullable
     ExecutionKind kind;
 
+    @Nullable
+    List<Breakpoint> breakpoints;
+
     /**
      * Factory method for constructing a new {@link Execution} object for the given {@link Flow}.
      *
@@ -221,7 +225,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
         );
     }
 
@@ -247,7 +252,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
         );
     }
 
@@ -286,7 +292,34 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
+        );
+    }
+
+    public Execution withBreakpoints(List<Breakpoint> newBreakpoints) {
+        return new Execution(
+            this.tenantId,
+            this.id,
+            this.namespace,
+            this.flowId,
+            this.flowRevision,
+            this.taskRunList,
+            this.inputs,
+            this.outputs,
+            this.labels,
+            this.variables,
+            this.state,
+            this.parentId,
+            this.originalId,
+            this.trigger,
+            this.deleted,
+            this.metadata,
+            this.scheduleDate,
+            this.traceParent,
+            this.fixtures,
+            this.kind,
+            newBreakpoints
         );
     }
 
@@ -312,7 +345,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
         );
     }
 
@@ -824,7 +858,7 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param e the current exception
      * @return the {@link ILoggingEvent} waited to generate {@link LogEntry}
      */
-    public static ILoggingEvent loggingEventFromException(Exception e) {
+    public static ILoggingEvent loggingEventFromException(Throwable e) {
         LoggingEvent loggingEvent = new LoggingEvent();
         loggingEvent.setLevel(ch.qos.logback.classic.Level.ERROR);
         loggingEvent.setThrowableProxy(new ThrowableProxy(e));

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

@@ -3,8 +3,9 @@ package io.kestra.core.models.executions;
 /**
  * Describe the kind of execution:
  * - TEST: created by a test
+ * - PLAYGROUND: created by a playground
  * - NORMAL: anything else, for backward compatibility NORMAL is not persisted but null is used instead
  */
 public enum ExecutionKind {
-    NORMAL, TEST
+    NORMAL, TEST, PLAYGROUND
 }

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

@@ -7,6 +7,7 @@ import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.utils.IdUtils;
 import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.annotation.Nullable;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import lombok.*;
@@ -62,6 +63,11 @@ public class TaskRun implements TenantInterface {
     @With
     Boolean dynamic;
 
+    // Set it to true to force execution even if the execution is killed
+    @Nullable
+    @With
+    Boolean forceExecution;
+
     @Deprecated
     public void setItems(String items) {
         // no-op for backward compatibility
@@ -81,7 +87,8 @@ public class TaskRun implements TenantInterface {
             this.outputs,
             this.state.withState(state),
             this.iteration,
-            this.dynamic
+            this.dynamic,
+            this.forceExecution
         );
     }
 
@@ -99,7 +106,8 @@ public class TaskRun implements TenantInterface {
             this.outputs,
             newState,
             this.iteration,
-            this.dynamic
+            this.dynamic,
+            this.forceExecution
         );
     }
 
@@ -121,7 +129,8 @@ public class TaskRun implements TenantInterface {
             this.outputs,
             this.state.withState(State.Type.FAILED),
             this.iteration,
-            this.dynamic
+            this.dynamic,
+            this.forceExecution
         );
     }
 
@@ -245,7 +254,7 @@ public class TaskRun implements TenantInterface {
      * @return The next retry date, null if maxAttempt || maxDuration is reached
      */
     public Instant nextRetryDate(AbstractRetry retry, Execution execution) {
-        if (retry.getMaxAttempt() != null && execution.getMetadata().getAttemptNumber() >= retry.getMaxAttempt()) {
+        if (retry.getMaxAttempts() != null && execution.getMetadata().getAttemptNumber() >= retry.getMaxAttempts()) {
 
             return null;
         }
@@ -265,7 +274,7 @@ public class TaskRun implements TenantInterface {
      * @return The next retry date, null if maxAttempt || maxDuration is reached
      */
     public Instant nextRetryDate(AbstractRetry retry) {
-        if (this.attempts == null || this.attempts.isEmpty() || (retry.getMaxAttempt() != null && this.attemptNumber() >= retry.getMaxAttempt())) {
+        if (this.attempts == null || this.attempts.isEmpty() || (retry.getMaxAttempts() != null && this.attemptNumber() >= retry.getMaxAttempts())) {
 
             return null;
         }

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

@@ -3,6 +3,8 @@ package io.kestra.core.models.flows;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import io.kestra.core.models.Label;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
 import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import io.swagger.v3.oas.annotations.Hidden;
@@ -60,6 +62,9 @@ public abstract class AbstractFlow implements FlowInterface {
     @Schema(implementation = Object.class, oneOf = {List.class, Map.class})
     List<Label> labels;
 
+    @Schema(additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
     Map<String, Object> variables;
 
+    @Valid
+    private WorkerGroup workerGroup;
 }

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

@@ -19,7 +19,6 @@ import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.serializers.JacksonMapper;
-import io.kestra.core.services.FlowService;
 import io.kestra.core.utils.ListUtils;
 import io.kestra.core.validations.FlowValidation;
 import io.micronaut.core.annotation.Introspected;
@@ -30,8 +29,6 @@ import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotEmpty;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.*;
 import java.util.stream.Collectors;
@@ -70,6 +67,8 @@ public class Flow extends AbstractFlow implements HasUID {
 
     @Valid
     @NotEmpty
+
+    @Schema(additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
     List<Task> tasks;
 
     @Valid
@@ -187,19 +186,32 @@ public class Flow extends AbstractFlow implements HasUID {
             .toList();
     }
 
-    public List<Task> allErrorsWithChilds() {
+    public List<Task> allErrorsWithChildren() {
         var allErrors = allTasksWithChilds().stream()
             .filter(task -> task.isFlowable() && ((FlowableTask<?>) task).getErrors() != null)
             .flatMap(task -> ((FlowableTask<?>) task).getErrors().stream())
             .collect(Collectors.toCollection(ArrayList::new));
 
-        if (this.getErrors() != null && !this.getErrors().isEmpty()) {
+        if (!ListUtils.isEmpty(this.getErrors())) {
             allErrors.addAll(this.getErrors());
         }
 
         return allErrors;
     }
 
+    public List<Task> allFinallyWithChildren() {
+        var allFinally = allTasksWithChilds().stream()
+            .filter(task -> task.isFlowable() && ((FlowableTask<?>) task).getFinally() != null)
+            .flatMap(task -> ((FlowableTask<?>) task).getFinally().stream())
+            .collect(Collectors.toCollection(ArrayList::new));
+
+        if (!ListUtils.isEmpty(this.getFinally())) {
+            allFinally.addAll(this.getFinally());
+        }
+
+        return allFinally;
+    }
+
     public Task findParentTasksByTaskId(String taskId) {
         return allTasksWithChilds()
             .stream()

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

@@ -11,6 +11,7 @@ import io.kestra.core.models.HasUID;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.flows.sla.SLA;
+import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.serializers.JacksonMapper;
 
 import java.util.AbstractMap;
@@ -42,6 +43,8 @@ public interface FlowInterface extends FlowId, DeletedInterface, TenantInterface
 
     Map<String, Object> getVariables();
 
+    WorkerGroup getWorkerGroup();
+
     default Concurrency getConcurrency() {
         return null;
     }

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

@@ -168,6 +168,11 @@ public class State {
         return this.current.isPaused();
     }
 
+    @JsonIgnore
+    public boolean isBreakpoint() {
+        return this.current.isBreakpoint();
+    }
+
     @JsonIgnore
     public boolean isRetrying() {
         return this.current.isRetrying();
@@ -216,7 +221,8 @@ public class State {
         QUEUED,
         RETRYING,
         RETRIED,
-        SKIPPED;
+        SKIPPED,
+        BREAKPOINT;
 
         public boolean isTerminated() {
             return this == Type.FAILED || this == Type.WARNING || this == Type.SUCCESS || this == Type.KILLED || this == Type.CANCELLED || this == Type.RETRIED || this == Type.SKIPPED;
@@ -242,6 +248,10 @@ public class State {
             return this == Type.PAUSED;
         }
 
+        public boolean isBreakpoint() {
+            return this == Type.BREAKPOINT;
+        }
+
         public boolean isRetrying() {
             return this == Type.RETRYING || this == Type.RETRIED;
         }

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

@@ -21,6 +21,7 @@ public class FileInput extends Input<URI> {
     private static final String DEFAULT_EXTENSION = ".upl";
 
     @Builder.Default
+    @Deprecated(since = "0.24", forRemoval = true)
     public String extension = DEFAULT_EXTENSION;
 
     @Override

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

@@ -108,7 +108,7 @@ public class MultiselectInput extends Input<List<String>> implements ItemTypeInt
     private List<String> renderExpressionValues(final Function<String, Object> renderer) {
         Object result;
         try {
-            result = renderer.apply(expression);
+            result = renderer.apply(expression.trim());
         } catch (Exception e) {
             throw ManualConstraintViolation.toConstraintViolationException(
                 "Cannot render 'expression'. Cause: " + e.getMessage(),

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

@@ -86,7 +86,7 @@ public class SelectInput extends Input<String> implements RenderableInput {
     private List<String> renderExpressionValues(final Function<String, Object> renderer) {
         Object result;
         try {
-            result = renderer.apply(expression);
+            result = renderer.apply(expression.trim());
         } catch (Exception e) {
             throw ManualConstraintViolation.toConstraintViolationException(
                 "Cannot render 'expression'. Cause: " + e.getMessage(),

core/src/main/java/io/kestra/core/models/property/Data.java

@@ -136,7 +136,7 @@ public class Data {
                 Structured data items can be defined in the following ways:
                 - A single item as a map (a document).
                 - A list of items as a list of maps (a list of documents).
-                - A URI, supported schemes are `kestra` for internal storage files, and `file` for host local files.
+                - A URI, supported schemes are `kestra` for internal storage files, `file` for host local files, and `nsfile` for namespace files.
                 - A JSON String that will then be serialized either as a single item or a list of items.""";
 
         @Schema(

core/src/main/java/io/kestra/core/models/property/URIFetcher.java

@@ -1,15 +1,14 @@
 package io.kestra.core.models.property;
 
-import io.kestra.core.runners.DefaultRunContext;
+import io.kestra.core.runners.LocalPath;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.storages.Namespace;
 import io.kestra.core.storages.StorageContext;
 
-import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.nio.file.Path;
-import java.util.Collections;
 import java.util.List;
 
 /**
@@ -17,8 +16,7 @@ import java.util.List;
  * It supports reading from the following schemes: {@link #SUPPORTED_SCHEMES}.
  */
 public class URIFetcher {
-    private static final String FILE_SCHEME = "file";
-    private static final List<String> SUPPORTED_SCHEMES = List.of(StorageContext.KESTRA_SCHEME, FILE_SCHEME);
+    private static final List<String> SUPPORTED_SCHEMES = List.of(StorageContext.KESTRA_SCHEME, LocalPath.FILE_SCHEME, Namespace.NAMESPACE_FILE_SCHEME);
 
     private final URI uri;
 
@@ -68,6 +66,14 @@ public class URIFetcher {
         return SUPPORTED_SCHEMES.stream().anyMatch(scheme -> uri.startsWith(scheme + "://"));
     }
 
+    /**
+     * Whether the URI is supported by the Fetcher.
+     * A supported URI is a URI which scheme is one of the {@link #SUPPORTED_SCHEMES}.
+     */
+    public static boolean supports(URI uri) {
+        return uri.getScheme() != null && SUPPORTED_SCHEMES.contains(uri.getScheme());
+    }
+
     /**
      * Fetch the resource pointed by this SmartURI
      *
@@ -82,23 +88,11 @@ public class URIFetcher {
         // we need to first check the protocol, then create one reader by protocol
         return switch (uri.getScheme()) {
             case StorageContext.KESTRA_SCHEME -> runContext.storage().getFile(uri);
-            case FILE_SCHEME -> {
-                Path path = Path.of(uri).toRealPath(); // toRealPath() will protect about path traversal issues
-                Path workingDirectory = runContext.workingDir().path();
-                if (!path.startsWith(workingDirectory)) {
-                    // we need to check that it's on an allowed path
-                    List<String> globalAllowedPaths = ((DefaultRunContext) runContext).getApplicationContext().getProperty("kestra.plugins.allowed-paths", List.class, Collections.emptyList());
-                    if (globalAllowedPaths.stream().noneMatch(path::startsWith)) {
-                        // if not globally allowed, we check it's allowed for this specific plugin
-                        List<String> pluginAllowedPaths = (List<String>) runContext.pluginConfiguration("allowed-paths").orElse(Collections.emptyList());
-                        if (pluginAllowedPaths.stream().noneMatch(path::startsWith)) {
-                            throw new SecurityException("The path " + path + " is not authorized. " +
-                                "Only files inside the working directory are allowed by default, other path must be allowed either globally inside the Kestra configuration using the `kestra.plugins.allowed-paths` property, " +
-                                "or by plugin using the `allowed-paths` plugin configuration.");
-                        }
-                    }
-                }
-                yield new FileInputStream(path.toFile());
+            case LocalPath.FILE_SCHEME -> runContext.localPath().get(uri);
+            case Namespace.NAMESPACE_FILE_SCHEME -> {
+                var namespace = uri.getAuthority() == null ? runContext.storage().namespace() : runContext.storage().namespace(uri.getAuthority());
+                var nsFileUri = namespace.get(Path.of(uri.getPath())).uri();
+                yield runContext.storage().getFile(nsFileUri);
             }
             default -> throw new IllegalArgumentException("Scheme not supported: " + uri.getScheme());
         };

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

@@ -0,0 +1,20 @@
+package io.kestra.core.models.tasks;
+
+import io.micronaut.core.annotation.Introspected;
+import jakarta.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.Duration;
+
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@Introspected
+public class Cache {
+    @NotNull
+    private Boolean enabled;
+
+    private Duration ttl;
+}

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

@@ -7,7 +7,12 @@ import java.util.Map;
 
 public interface InputFilesInterface {
     @Schema(
-        title = "The files to create on the local filesystem. It can be a map or a JSON object.",
+        title = "The files to create on the working. It can be a map or a JSON object.",
+        description = """
+            Each file can be defined:
+            - Inline with its content
+            - As a URI, supported schemes are `kestra` for internal storage files, `file` for host local files, and `nsfile` for namespace files.
+            """,
         oneOf = {Map.class, String.class}
     )
     @PluginProperty(dynamic = true)

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

@@ -49,4 +49,10 @@ public class NamespaceFiles {
     )
     @Builder.Default
     private Property<FileExistComportment> ifExists = Property.ofValue(FileExistComportment.OVERWRITE);
+
+    @Schema(
+        title = "Whether to mount file into the root of the working directory, or create a folder per namespace"
+    )
+    @Builder.Default
+    private Property<Boolean> folderPerNamespace = Property.ofValue(false);
 }

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

@@ -11,6 +11,7 @@ import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.core.flow.WorkingDirectory;
 import jakarta.validation.Valid;
+import jakarta.validation.constraints.Size;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -28,6 +29,7 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 @Plugin
 abstract public class Task implements TaskInterface {
+    @Size(max = 256, message = "Task id must be at most 256 characters")
     protected String id;
 
     protected String type;
@@ -72,6 +74,10 @@ abstract public class Task implements TaskInterface {
     @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
     private boolean allowWarning = false;
 
+    @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
+    @Valid
+    private Cache taskCache;
+
     public Optional<Task> findById(String id) {
         if (this.getId().equals(id)) {
             return Optional.of(this);

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

@@ -1,6 +1,5 @@
 package io.kestra.core.models.tasks;
 
-import io.kestra.core.validations.WorkerGroupValidation;
 import io.micronaut.core.annotation.Introspected;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
@@ -10,7 +9,6 @@ import lombok.NoArgsConstructor;
 @NoArgsConstructor
 @AllArgsConstructor
 @Introspected
-@WorkerGroupValidation
 public class WorkerGroup {
 
     private String key;

core/src/main/java/io/kestra/core/models/tasks/retrys/AbstractRetry.java

@@ -29,8 +29,18 @@ public abstract class AbstractRetry {
 
     private Duration maxDuration;
 
+    @Deprecated(forRemoval = true)
+    public Integer getMaxAttempt() {
+        return maxAttempts;
+    }
+
+    @Deprecated(forRemoval = true)
+    public void setMaxAttempt(@Min(1) Integer maxAttempt) {
+        this.maxAttempts = maxAttempt;
+    }
+
     @Min(1)
-    private Integer maxAttempt;
+    private Integer maxAttempts;
 
     @Builder.Default
     private Boolean warningOnRetry = false;
@@ -46,8 +56,8 @@ public abstract class AbstractRetry {
             builder.withMaxDuration(maxDuration);
         }
 
-        if (this.maxAttempt != null) {
-            builder.withMaxAttempts(this.maxAttempt);
+        if (this.maxAttempts != null) {
+            builder.withMaxAttempts(this.maxAttempts);
         }
         return builder;
     }

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

@@ -100,7 +100,7 @@ abstract public class PluginUtilsService {
     @SuppressWarnings("unchecked")
     public static Map<String, String> transformInputFiles(RunContext runContext, Map<String, Object> additionalVars, @NotNull Object inputFiles) throws IllegalVariableEvaluationException, JsonProcessingException {
         if (inputFiles instanceof Map) {
-            Map<String, String> castedInputFiles = (Map<String, String>) ((Map<?, ?>) inputFiles);
+            Map<String, String> castedInputFiles = (Map<String, String>) inputFiles;
             Map<String, String> nullFilteredInputFiles = new HashMap<>();
             castedInputFiles.forEach((key, val) -> {
                 if (val != null) {
@@ -110,7 +110,6 @@ abstract public class PluginUtilsService {
             return runContext.renderMap(nullFilteredInputFiles, additionalVars);
         } else if (inputFiles instanceof String inputFileString) {
 
-
             return JacksonMapper.ofJson(false).readValue(
                 runContext.render(inputFileString, additionalVars),
                 MAP_TYPE_REFERENCE

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

@@ -30,7 +30,7 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
  * Helper class for task runners and script tasks.
  */
 public final class ScriptService {
-    private static final Pattern INTERNAL_STORAGE_PATTERN = Pattern.compile("(kestra:\\/\\/[-a-zA-Z0-9%._\\+~#=/]*)");
+    private static final Pattern INTERNAL_STORAGE_PATTERN = Pattern.compile("(kestra:\\/\\/[-\\p{Alnum}._\\+~#=/]*)", Pattern.UNICODE_CHARACTER_CLASS);
 
     // These are the three common additional variables task runners must provide for variable rendering.
     public static final String VAR_WORKING_DIR = "workingDir";

core/src/main/java/io/kestra/core/models/topologies/FlowNode.java

@@ -1,7 +1,6 @@
 package io.kestra.core.models.topologies;
 
 import io.kestra.core.models.TenantInterface;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowInterface;
 import io.swagger.v3.oas.annotations.Hidden;
 import lombok.AllArgsConstructor;
@@ -11,6 +10,8 @@ import lombok.experimental.SuperBuilder;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 
+import java.util.Objects;
+
 @Getter
 @AllArgsConstructor
 @SuperBuilder(toBuilder = true)
@@ -26,6 +27,18 @@ public class FlowNode implements TenantInterface {
 
     String id;
 
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        FlowNode flowNode = (FlowNode) o;
+        return Objects.equals(uid, flowNode.uid);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(uid);
+    }
+
     public static FlowNode of(FlowInterface flow) {
         return FlowNode.builder()
             .uid(flow.uidWithoutRevision())

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

@@ -78,6 +78,10 @@ abstract public class AbstractTrigger implements TriggerInterface {
     @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
     private boolean logToFile = false;
 
+    @Builder.Default
+    @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
+    private boolean failOnTriggerError = false;
+
     /**
      * For backward compatibility: we rename minLogLevel to logLevel.
      * @deprecated use {@link #logLevel} instead

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

@@ -8,6 +8,7 @@ import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.Setter;
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
@@ -20,6 +21,7 @@ import java.util.List;
 @NoArgsConstructor
 @Introspected
 public class TriggerContext {
+    @Setter
     @Pattern(regexp = "^[a-z0-9][a-z0-9_-]")
     private String tenantId;
 

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

@@ -119,6 +119,7 @@ public abstract class TriggerService {
             .id(id)
             .namespace(context.getNamespace())
             .flowId(context.getFlowId())
+            .tenantId(context.getTenantId())
             .flowRevision(flowRevision)
             .state(new State())
             .trigger(executionTrigger)

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

@@ -249,9 +249,9 @@ public class PluginScanner {
     }
 
     private static void addGuides(Path root, List<String> guides) throws IOException {
-        try (var stream = Files.walk(root, 1)) {
+        try (var stream = Files.walk(root)) { // remove depth limit to walk recursively
             stream
-                .skip(1) // first element is the root element
+                .filter(Files::isRegularFile)
                 .sorted(Comparator.comparing(path -> path.getName(path.getParent().getNameCount()).toString()))
                 .forEach(guide -> {
                     var guideName = guide.getName(guide.getParent().getNameCount()).toString();

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

@@ -33,6 +33,20 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @EqualsAndHashCode
 @Builder
 public class RegisteredPlugin {
+    public static final String TASKS_GROUP_NAME = "tasks";
+    public static final String TRIGGERS_GROUP_NAME = "triggers";
+    public static final String CONDITIONS_GROUP_NAME = "conditions";
+    public static final String STORAGES_GROUP_NAME = "storages";
+    public static final String SECRETS_GROUP_NAME = "secrets";
+    public static final String TASK_RUNNERS_GROUP_NAME = "task-runners";
+    public static final String APPS_GROUP_NAME = "apps";
+    public static final String APP_BLOCKS_GROUP_NAME = "app-blocks";
+    public static final String CHARTS_GROUP_NAME = "charts";
+    public static final String DATA_FILTERS_GROUP_NAME = "data-filters";
+    public static final String DATA_FILTERS_KPI_GROUP_NAME = "data-filters-kpi";
+    public static final String LOG_EXPORTERS_GROUP_NAME = "log-exporters";
+    public static final String ADDITIONAL_PLUGINS_GROUP_NAME = "additional-plugins";
+
     private final ExternalPlugin externalPlugin;
     private final Manifest manifest;
     private final ClassLoader classLoader;
@@ -160,19 +174,19 @@ public class RegisteredPlugin {
     public Map<String, List<Class>> allClassGrouped() {
         Map<String, List<Class>> result = new HashMap<>();
 
-        result.put("tasks", Arrays.asList(this.getTasks().toArray(Class[]::new)));
-        result.put("triggers", Arrays.asList(this.getTriggers().toArray(Class[]::new)));
-        result.put("conditions", Arrays.asList(this.getConditions().toArray(Class[]::new)));
-        result.put("storages", Arrays.asList(this.getStorages().toArray(Class[]::new)));
-        result.put("secrets", Arrays.asList(this.getSecrets().toArray(Class[]::new)));
-        result.put("task-runners", Arrays.asList(this.getTaskRunners().toArray(Class[]::new)));
-        result.put("apps", Arrays.asList(this.getApps().toArray(Class[]::new)));
-        result.put("app-blocks", Arrays.asList(this.getAppBlocks().toArray(Class[]::new)));
-        result.put("charts", Arrays.asList(this.getCharts().toArray(Class[]::new)));
-        result.put("data-filters", Arrays.asList(this.getDataFilters().toArray(Class[]::new)));
-        result.put("data-filters-kpi", Arrays.asList(this.getDataFiltersKPI().toArray(Class[]::new)));
-        result.put("log-exporters", Arrays.asList(this.getLogExporters().toArray(Class[]::new)));
-        result.put("additional-plugins", Arrays.asList(this.getAdditionalPlugins().toArray(Class[]::new)));
+        result.put(TASKS_GROUP_NAME, Arrays.asList(this.getTasks().toArray(Class[]::new)));
+        result.put(TRIGGERS_GROUP_NAME, Arrays.asList(this.getTriggers().toArray(Class[]::new)));
+        result.put(CONDITIONS_GROUP_NAME, Arrays.asList(this.getConditions().toArray(Class[]::new)));
+        result.put(STORAGES_GROUP_NAME, Arrays.asList(this.getStorages().toArray(Class[]::new)));
+        result.put(SECRETS_GROUP_NAME, Arrays.asList(this.getSecrets().toArray(Class[]::new)));
+        result.put(TASK_RUNNERS_GROUP_NAME, Arrays.asList(this.getTaskRunners().toArray(Class[]::new)));
+        result.put(APPS_GROUP_NAME, Arrays.asList(this.getApps().toArray(Class[]::new)));
+        result.put(APP_BLOCKS_GROUP_NAME, Arrays.asList(this.getAppBlocks().toArray(Class[]::new)));
+        result.put(CHARTS_GROUP_NAME, Arrays.asList(this.getCharts().toArray(Class[]::new)));
+        result.put(DATA_FILTERS_GROUP_NAME, Arrays.asList(this.getDataFilters().toArray(Class[]::new)));
+        result.put(DATA_FILTERS_KPI_GROUP_NAME, Arrays.asList(this.getDataFiltersKPI().toArray(Class[]::new)));
+        result.put(LOG_EXPORTERS_GROUP_NAME, Arrays.asList(this.getLogExporters().toArray(Class[]::new)));
+        result.put(ADDITIONAL_PLUGINS_GROUP_NAME, Arrays.asList(this.getAdditionalPlugins().toArray(Class[]::new)));
 
         return result;
     }

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

@@ -24,6 +24,8 @@ public interface DashboardRepositoryInterface {
 
     List<Dashboard> findAll(String tenantId);
 
+    List<Dashboard> findAllWithNoAcl(String tenantId);
+
     default Dashboard save(Dashboard dashboard, String source) {
         return this.save(null, dashboard, source);
     }

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

@@ -5,7 +5,6 @@ import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.executions.statistics.DailyExecutionStatistics;
 import io.kestra.core.models.executions.statistics.ExecutionCount;
-import io.kestra.core.models.executions.statistics.ExecutionCountStatistics;
 import io.kestra.core.models.executions.statistics.Flow;
 import io.kestra.core.models.flows.FlowScope;
 import io.kestra.core.models.flows.State;
@@ -130,29 +129,6 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
         boolean isTaskRun
     );
 
-    List<Execution> lastExecutions(
-        @Nullable String tenantId,
-        @Nullable List<FlowFilter> flows
-    );
-
-    Map<String, Map<String, List<DailyExecutionStatistics>>> dailyGroupByFlowStatistics(
-        @Nullable String query,
-        @Nullable String tenantId,
-        @Nullable String namespace,
-        @Nullable String flowId,
-        @Nullable List<FlowFilter> flows,
-        @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate,
-        boolean groupByNamespaceOnly
-    );
-
-    Map<String, ExecutionCountStatistics> executionCountsGroupedByNamespace(
-        @Nullable String tenantId,
-        @Nullable String namespace,
-        @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate
-    );
-
     @Getter
     @SuperBuilder
     @NoArgsConstructor
@@ -183,4 +159,9 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
         CHILD,
         MAIN
     }
+
+    List<Execution> lastExecutions(
+        @Nullable String tenantId,
+        @Nullable List<FlowFilter> flows
+    );
 }

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

@@ -3,18 +3,12 @@ package io.kestra.core.repositories;
 import io.kestra.core.models.QueryFilter;
 import io.kestra.core.models.SearchResult;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.FlowForExecution;
-import io.kestra.core.models.flows.FlowInterface;
-import io.kestra.core.models.flows.FlowScope;
-import io.kestra.core.models.flows.FlowWithSource;
-import io.kestra.core.models.flows.GenericFlow;
+import io.kestra.core.models.flows.*;
 import io.micronaut.data.model.Pageable;
-
 import jakarta.annotation.Nullable;
 import jakarta.validation.ConstraintViolationException;
+
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 public interface FlowRepositoryInterface {
@@ -109,6 +103,8 @@ public interface FlowRepositoryInterface {
 
     List<FlowWithSource> findAllWithSource(String tenantId);
 
+    List<FlowWithSource> findAllWithSourceWithNoAcl(String tenantId);
+
     List<Flow> findAllForAllTenants();
 
     List<FlowWithSource> findAllWithSourceForAllTenants();
@@ -121,14 +117,6 @@ public interface FlowRepositoryInterface {
      */
     int count(@Nullable  String tenantId);
 
-    /**
-     * Counts the total number of flows for the given namespace.
-     *
-     * @param tenantId the tenant ID.
-     * @return The count.
-     */
-    int countForNamespace(@Nullable  String tenantId, @Nullable String namespace);
-
     List<Flow> findByNamespace(String tenantId, String namespace);
 
     List<Flow> findByNamespacePrefix(String tenantId, String namespacePrefix);

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

@@ -10,5 +10,7 @@ public interface FlowTopologyRepositoryInterface {
 
     List<FlowTopology> findByNamespace(String tenantId, String namespace);
 
+    List<FlowTopology> findAll(String tenantId);
+
     FlowTopology save(FlowTopology flowTopology);
 }

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

@@ -3,16 +3,14 @@ package io.kestra.core.repositories;
 import io.kestra.core.models.QueryFilter;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
-import io.kestra.core.models.executions.statistics.LogStatistics;
-import io.kestra.core.utils.DateUtils;
 import io.kestra.plugin.core.dashboard.data.Logs;
 import io.micronaut.data.model.Pageable;
 import jakarta.annotation.Nullable;
 import org.slf4j.event.Level;
+import reactor.core.publisher.Flux;
 
 import java.time.ZonedDateTime;
 import java.util.List;
-import reactor.core.publisher.Flux;
 
 public interface LogRepositoryInterface extends SaveRepositoryInterface<LogEntry>, QueryBuilderInterface<Logs.Fields> {
     /**
@@ -90,17 +88,6 @@ public interface LogRepositoryInterface extends SaveRepositoryInterface<LogEntry
 
     Flux<LogEntry> findAllAsync(@Nullable String tenantId);
 
-    List<LogStatistics> statistics(
-        @Nullable String query,
-        @Nullable String tenantId,
-        @Nullable String namespace,
-        @Nullable String flowId,
-        @Nullable Level minLevel,
-        @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate,
-        @Nullable DateUtils.GroupType groupBy
-    );
-
     LogEntry save(LogEntry log);
 
     Integer purge(Execution execution);
@@ -109,5 +96,5 @@ public interface LogRepositoryInterface extends SaveRepositoryInterface<LogEntry
 
     void deleteByQuery(String tenantId, String namespace, String flowId, String triggerId);
 
-    int deleteByQuery(String tenantId, String namespace, String flowId, List<Level> logLevels, ZonedDateTime startDate, ZonedDateTime endDate);
+    int deleteByQuery(String tenantId, String namespace, String flowId, String executionId, List<Level> logLevels, ZonedDateTime startDate, ZonedDateTime endDate);
 }

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

@@ -12,6 +12,8 @@ public interface TemplateRepositoryInterface {
 
     List<Template> findAll(String tenantId);
 
+    List<Template> findAllWithNoAcl(String tenantId);
+
     List<Template> findAllForAllTenants();
 
     ArrayListTotal<Template> find(

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

@@ -41,15 +41,6 @@ public interface TriggerRepositoryInterface extends QueryBuilderInterface<Trigge
      */
     int count(@Nullable String tenantId);
 
-    /**
-     * Counts the total number of triggers for the given namespace.
-     *
-     * @param tenantId  the tenant of the triggers
-     * @param namespace the namespace
-     * @return The count.
-     */
-    int countForNamespace(@Nullable String tenantId, @Nullable String namespace);
-
     /**
      * Find all triggers that match the query, return a flux of triggers
      * as the search is not paginated

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

@@ -15,7 +15,6 @@ import io.kestra.core.storages.Storage;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.storages.kv.KVStore;
 import io.kestra.core.utils.ListUtils;
-import io.kestra.core.utils.MapUtils;
 import io.kestra.core.utils.VersionProvider;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.core.annotation.Introspected;
@@ -56,6 +55,7 @@ public class DefaultRunContext extends RunContext {
     private Optional<String> secretKey;
     private WorkingDir workingDir;
     private Validator validator;
+    private LocalPath localPath;
 
     private Map<String, Object> variables;
     private List<AbstractMetricEntry<?>> metrics = new ArrayList<>();
@@ -153,6 +153,7 @@ public class DefaultRunContext extends RunContext {
             this.kvStoreService = applicationContext.getBean(KVStoreService.class);
             this.secretKey = applicationContext.getProperty("kestra.encryption.secret-key", String.class);
             this.validator = applicationContext.getBean(Validator.class);
+            this.localPath = applicationContext.getBean(LocalPathFactory.class).createLocalPath(this);
         }
     }
 
@@ -573,6 +574,11 @@ public class DefaultRunContext extends RunContext {
         return isInitialized.get();
     }
 
+    @Override
+    public LocalPath localPath() {
+        return localPath;
+    }
+
     /**
      * Builder class for constructing new {@link DefaultRunContext} objects.
      */

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

@@ -30,5 +30,5 @@ public class ExecutionRunning {
         return IdUtils.fromPartsAndSeparator('|', this.tenantId, this.namespace, this.flowId, this.execution.getId());
     }
 
-    public enum ConcurrencyState { CREATED, RUNNING, QUEUED }
+    public enum ConcurrencyState { CREATED, RUNNING, QUEUED, CANCELLED, FAILED }
 }

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

@@ -85,7 +85,8 @@ public class Executor {
     }
 
     public Boolean canBeProcessed() {
-        return !(this.getException() != null || this.getFlow() == null || this.getFlow() instanceof FlowWithException || this.getFlow().getTasks() == null || this.getExecution().isDeleted() || this.getExecution().getState().isPaused());
+        return !(this.getException() != null || this.getFlow() == null || this.getFlow() instanceof FlowWithException || this.getFlow().getTasks() == null ||
+            this.getExecution().isDeleted() || this.getExecution().getState().isPaused() || this.getExecution().getState().isBreakpoint());
     }
 
     public Executor withFlow(FlowWithSource flow) {

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

@@ -1,5 +1,6 @@
 package io.kestra.core.runners;
 
+import io.kestra.core.debug.Breakpoint;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.Label;
@@ -18,6 +19,7 @@ import io.kestra.core.storages.StorageContext;
 import io.kestra.core.test.flow.TaskFixture;
 import io.kestra.core.trace.propagation.RunContextTextMapSetter;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.MapUtils;
 import io.kestra.core.utils.TruthUtils;
 import io.kestra.plugin.core.flow.LoopUntil;
 import io.kestra.plugin.core.flow.Pause;
@@ -260,8 +262,10 @@ public class ExecutorService {
                 // Compute outputs for the parent Flowable task if a terminated state was resolved
                 if (workerTaskResult.getTaskRun().getState().isTerminated()) {
                     try {
+                        // as flowable tasks can save outputs during iterative execution, we must merge the maps here
                         Output outputs = flowableParent.outputs(runContext);
-                        Variables variables = variablesService.of(StorageContext.forTask(workerTaskResult.getTaskRun()), outputs);
+                        Map<String, Object> outputMap = MapUtils.merge(workerTaskResult.getTaskRun().getOutputs(), outputs == null ? null : outputs.toMap());
+                        Variables variables = variablesService.of(StorageContext.forTask(workerTaskResult.getTaskRun()), outputMap);
                         return Optional.of(new WorkerTaskResult(workerTaskResult
                             .getTaskRun()
                             .withOutputs(variables)
@@ -735,6 +739,7 @@ public class ExecutorService {
         List<TaskRun> afterExecutionNexts = FlowableUtils.resolveSequentialNexts(executor.getExecution(), afterExecutionResolvedTasks)
             .stream()
             .map(throwFunction(NextTaskRun::getTaskRun))
+            .map(taskRun -> taskRun.withForceExecution(true)) // forceExecution so it would be executed even if the execution is killed
             .toList();
         if (!afterExecutionNexts.isEmpty()) {
             return executor.withTaskRun(afterExecutionNexts, "handleAfterExecution ");
@@ -818,7 +823,7 @@ public class ExecutorService {
                         .executionKind(executor.getExecution().getKind())
                         .build();
                     // Get worker group
-                    Optional<WorkerGroup> workerGroup = workerGroupService.resolveGroupFromJob(workerTask);
+                    Optional<WorkerGroup> workerGroup = workerGroupService.resolveGroupFromJob(executor.getFlow(), workerTask);
                     if (workerGroup.isPresent()) {
                         // Check if the worker group exist
                         String tenantId = executor.getFlow().getTenantId();
@@ -887,13 +892,38 @@ public class ExecutorService {
             this.addWorkerTaskResults(executor, workerTaskResults);
         }
 
-
         if (workerTasks.isEmpty() || hasMockedWorkerTask) {
             return executor;
         }
 
         Executor executorToReturn = executor;
 
+        // suspend on breakpoint: if a breakpoint is for a CREATED taskrun, set the execution state to BREAKPOINT and ends here
+        if (!ListUtils.isEmpty(executor.getExecution().getBreakpoints())) {
+            List<Breakpoint> breakpoints = executor.getExecution().getBreakpoints();
+            if (executor.getExecution()
+                .getTaskRunList()
+                .stream()
+                .anyMatch(taskRun -> shouldSuspend(taskRun, breakpoints))
+            ) {
+                List<TaskRun> newTaskRuns = executor.getExecution().getTaskRunList().stream().map(
+                    taskRun -> {
+                        if (shouldSuspend(taskRun, breakpoints)) {
+                            return taskRun.withState(State.Type.BREAKPOINT);
+                        }
+                        return taskRun;
+                    }
+                ).toList();
+                Execution newExecution = executor.getExecution().withTaskRunList(newTaskRuns).withState(State.Type.BREAKPOINT);
+                executorToReturn = executorToReturn.withExecution(newExecution, "handleBreakpoint");
+                logService.logExecution(
+                    newExecution,
+                    Level.INFO,
+                    "Flow is suspended at a breakpoint."
+                );
+            }
+        }
+
         // Ends FAILED or CANCELLED task runs by creating worker task results
         List<WorkerTask> endedTasks = workerTasks.get(true);
         if (endedTasks != null && !endedTasks.isEmpty()) {
@@ -907,7 +937,7 @@ public class ExecutorService {
 
         // Send other TaskRun to the worker (create worker tasks)
         List<WorkerTask> processingTasks = workerTasks.get(false);
-        if (processingTasks != null && !processingTasks.isEmpty()) {
+        if (processingTasks != null && !processingTasks.isEmpty() && !executor.getExecution().getState().isBreakpoint()) {
             executorToReturn = executorToReturn.withWorkerTasks(processingTasks, "handleWorkerTask");
 
             metricRegistry.counter(MetricRegistry.METRIC_EXECUTOR_TASKRUN_CREATED_COUNT, MetricRegistry.METRIC_EXECUTOR_TASKRUN_CREATED_COUNT_DESCRIPTION, metricRegistry.tags(executor.getExecution())).increment(processingTasks.size());
@@ -916,6 +946,11 @@ public class ExecutorService {
         return executorToReturn;
     }
 
+    private boolean shouldSuspend(TaskRun taskRun, List<Breakpoint> breakpoints) {
+        return taskRun.getState().getCurrent().isCreated() && breakpoints.stream()
+                .anyMatch(breakpoint -> taskRun.getTaskId().equals(breakpoint.getId()) && (breakpoint.getValue() == null || Objects.equals(taskRun.getValue(), breakpoint.getValue())));
+    }
+
     private Executor handleExecutableTask(final Executor executor) {
         List<SubflowExecution<?>> executions = new ArrayList<>();
         List<SubflowExecutionResult> subflowExecutionResults = new ArrayList<>();
@@ -1138,71 +1173,83 @@ public class ExecutorService {
     }
 
     public void log(Logger log, Boolean in, WorkerJob value) {
-        if (value instanceof WorkerTask workerTask) {
-            log.debug(
-                "{} {} : {}",
-                in ? "<< IN " : ">> OUT",
-                workerTask.getClass().getSimpleName(),
-                workerTask.getTaskRun().toStringState()
-            );
-        } else if (value instanceof WorkerTrigger workerTrigger) {
-            log.debug(
-                "{} {} : {}",
-                in ? "<< IN " : ">> OUT",
-                workerTrigger.getClass().getSimpleName(),
-                workerTrigger.getTriggerContext().uid()
-            );
+        if (log.isDebugEnabled()) { // taskRun().toStringState() is costly so we avoid calling it if not needed
+            if (value instanceof WorkerTask workerTask) {
+                log.debug(
+                    "{} {} : {}",
+                    in ? "<< IN " : ">> OUT",
+                    workerTask.getClass().getSimpleName(),
+                    workerTask.getTaskRun().toStringState()
+                );
+            } else if (value instanceof WorkerTrigger workerTrigger) {
+                log.debug(
+                    "{} {} : {}",
+                    in ? "<< IN " : ">> OUT",
+                    workerTrigger.getClass().getSimpleName(),
+                    workerTrigger.getTriggerContext().uid()
+                );
+            }
         }
     }
 
     public void log(Logger log, Boolean in, WorkerTaskResult value) {
-        log.debug(
-            "{} {} : {}",
-            in ? "<< IN " : ">> OUT",
-            value.getClass().getSimpleName(),
-            value.getTaskRun().toStringState()
-        );
+        if (log.isDebugEnabled()) { // taskRun().toStringState() is costly so we avoid calling it if not needed
+            log.debug(
+                "{} {} : {}",
+                in ? "<< IN " : ">> OUT",
+                value.getClass().getSimpleName(),
+                value.getTaskRun().toStringState()
+            );
+        }
     }
 
     public void log(Logger log, Boolean in, SubflowExecutionResult value) {
-        log.debug(
-            "{} {} : {}",
-            in ? "<< IN " : ">> OUT",
-            value.getClass().getSimpleName(),
-            value.getParentTaskRun().toStringState()
-        );
+        if (log.isDebugEnabled()) { // taskRun().toStringState() is costly so we avoid calling it if not needed
+            log.debug(
+                "{} {} : {}",
+                in ? "<< IN " : ">> OUT",
+                value.getClass().getSimpleName(),
+                value.getParentTaskRun().toStringState()
+            );
+        }
     }
 
     public void log(Logger log, Boolean in, SubflowExecutionEnd value) {
-        log.debug(
-            "{} {} : {}",
-            in ? "<< IN " : ">> OUT",
-            value.getClass().getSimpleName(),
-            value.toStringState()
-        );
+        if (log.isDebugEnabled()) { // taskRun().toStringState() is costly so we avoid calling it if not needed
+            log.debug(
+                "{} {} : {}",
+                in ? "<< IN " : ">> OUT",
+                value.getClass().getSimpleName(),
+                value.toStringState()
+            );
+        }
     }
 
     public void log(Logger log, Boolean in, Execution value) {
-        log.debug(
-            "{} {} [key='{}']\n{}",
-            in ? "<< IN " : ">> OUT",
-            value.getClass().getSimpleName(),
-            value.getId(),
-            value.toStringState()
-        );
+        if (log.isDebugEnabled()) { // taskRun().toStringState() is costly so we avoid calling it if not needed
+            log.debug(
+                "{} {} [key='{}']\n{}",
+                in ? "<< IN " : ">> OUT",
+                value.getClass().getSimpleName(),
+                value.getId(),
+                value.toStringState()
+            );
+        }
     }
 
     public void log(Logger log, Boolean in, Executor value) {
-        log.debug(
-            "{} {} [key='{}', from='{}', offset='{}', crc32='{}']\n{}",
-            in ? "<< IN " : ">> OUT",
-            value.getClass().getSimpleName(),
-            value.getExecution().getId(),
-            value.getFrom(),
-            value.getOffset(),
-            value.getExecution().toCrc32State(),
-            value.getExecution().toStringState()
-        );
+        if (log.isDebugEnabled()) { // taskRun().toStringState() is costly so we avoid calling it if not needed
+            log.debug(
+                "{} {} [key='{}', from='{}', offset='{}', crc32='{}']\n{}",
+                in ? "<< IN " : ">> OUT",
+                value.getClass().getSimpleName(),
+                value.getExecution().getId(),
+                value.getFrom(),
+                value.getOffset(),
+                value.getExecution().toCrc32State(),
+                value.getExecution().toStringState()
+            );
+        }
     }
 
     public void log(Logger log, Boolean in, ExecutionKilledExecution value) {
@@ -1296,7 +1343,7 @@ public class ExecutorService {
             .state(ExecutionKilled.State.REQUESTED)
             .executionState(state)
             .executionId(execution.getId())
-            .isOnKillCascade(false) // TODO we may offer the choice to the user here
+            .isOnKillCascade(true)
             .tenantId(execution.getTenantId())
             .build()
         );

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

@@ -1,12 +1,13 @@
 package io.kestra.core.runners;
 
+import io.kestra.core.models.property.URIFetcher;
 import io.kestra.core.models.tasks.runners.PluginUtilsService;
+import io.kestra.core.serializers.FileSerde;
 import io.kestra.core.utils.IdUtils;
 import org.apache.commons.io.IOUtils;
 import org.slf4j.Logger;
 
-import java.io.File;
-import java.io.FileOutputStream;
+import java.io.*;
 import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -40,10 +41,13 @@ public abstract class FilesService {
                  }
 
                  if (input == null) {
-                    file.createNewFile();
+                    if(!file.createNewFile()) {
+                        throw new RuntimeException("Unable to create the file: " + file.getName());
+                    }
                  } else {
-                     if (input.startsWith("kestra://")) {
-                         try (var is = runContext.storage().getFile(URI.create(input));
+                     if (URIFetcher.supports(input)) {
+                         var uri = URIFetcher.of(input);
+                         try (var is = new BufferedInputStream(uri.fetch(runContext), FileSerde.BUFFER_SIZE);
                               var out = new FileOutputStream(file)) {
                              IOUtils.copyLarge(is, out);
                          }

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

@@ -17,6 +17,7 @@ import io.kestra.core.models.flows.Type;
 import io.kestra.core.models.flows.input.FileInput;
 import io.kestra.core.models.flows.input.InputAndValue;
 import io.kestra.core.models.flows.input.ItemTypeInterface;
+import io.kestra.core.models.property.URIFetcher;
 import io.kestra.core.models.tasks.common.EncryptedString;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.serializers.JacksonMapper;
@@ -137,19 +138,30 @@ public class FlowInputOutput {
             .publishOn(Schedulers.boundedElastic())
             .<AbstractMap.SimpleEntry<String, String>>handle((input, sink) -> {
                 if (input instanceof CompletedFileUpload fileUpload) {
+                    boolean oldStyleInput = false;
+                    if ("files".equals(fileUpload.getName())) {
+                        // we are maybe in an old-style usage of the input, let's check if there is an input named after the filename
+                        oldStyleInput = inputs.stream().anyMatch(i -> i.getId().equals(fileUpload.getFilename()));
+                    }
+                    if (oldStyleInput) {
+                        var runContext = runContextFactory.of(null, execution);
+                        runContext.logger().warn("Using a deprecated way to upload a FILE input. You must set the input 'id' as part name and set the name of the file using the regular 'filename' part attribute.");
+                    }
+                    String inputId = oldStyleInput ? fileUpload.getFilename() : fileUpload.getName();
+                    String fileName = oldStyleInput ? FileInput.findFileInputExtension(inputs, fileUpload.getFilename()) : fileUpload.getFilename();
+
                     if (!uploadFiles) {
-                        final String fileExtension = FileInput.findFileInputExtension(inputs, fileUpload.getFilename());
                         URI from = URI.create("kestra://" + StorageContext
-                            .forInput(execution, fileUpload.getFilename(), fileUpload.getFilename() + fileExtension)
+                            .forInput(execution, inputId, fileName)
                             .getContextStorageURI()
                         );
                         fileUpload.discard();
-                        sink.next(new AbstractMap.SimpleEntry<>(fileUpload.getFilename(), from.toString()));
+                        sink.next(new AbstractMap.SimpleEntry<>(inputId, from.toString()));
                     } else {
                         try {
-                            final String fileExtension = FileInput.findFileInputExtension(inputs, fileUpload.getFilename());
+                            final String fileExtension = FileInput.findFileInputExtension(inputs, fileName);
 
-                            String prefix = StringUtils.leftPad(fileUpload.getFilename() + "_", 3, "_");
+                            String prefix = StringUtils.leftPad(fileName + "_", 3, "_");
                             File tempFile = File.createTempFile(prefix, fileExtension);
                             try (var inputStream = fileUpload.getInputStream();
                                  var outputStream = new FileOutputStream(tempFile)) {
@@ -158,8 +170,8 @@ public class FlowInputOutput {
                                     sink.error(new KestraRuntimeException("Can't upload file: " + fileUpload.getFilename()));
                                     return;
                                 }
-                                URI from = storageInterface.from(execution, fileUpload.getFilename(), tempFile);
-                                sink.next(new AbstractMap.SimpleEntry<>(fileUpload.getFilename(), from.toString()));
+                                URI from = storageInterface.from(execution, inputId, fileName, tempFile);
+                                sink.next(new AbstractMap.SimpleEntry<>(inputId, from.toString()));
                             } finally {
                                 if (!tempFile.delete()) {
                                     tempFile.deleteOnExit();
@@ -425,10 +437,10 @@ public class FlowInputOutput {
                 case FILE -> {
                     URI uri = URI.create(current.toString().replace(File.separator, "/"));
 
-                    if (uri.getScheme() != null && uri.getScheme().equals("kestra")) {
+                    if (URIFetcher.supports(uri)) {
                         yield uri;
                     } else {
-                        yield storageInterface.from(execution, id, new File(current.toString()));
+                        yield storageInterface.from(execution, id, current.toString().substring(current.toString().lastIndexOf("/") + 1), new File(current.toString()));
                     }
                 }
                 case JSON -> JacksonMapper.toObject(current.toString());

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

@@ -0,0 +1,47 @@
+package io.kestra.core.runners;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.attribute.BasicFileAttributes;
+
+/**
+ * Get access to local paths of the host machine.
+ * <p>
+ * All methods of this class check allowed paths and protect against path traversal.
+ * All paths must be allowed via the {@link #ALLOWED_PATHS_CONFIG} configuration property or via plugin configuration.
+ */
+public interface LocalPath {
+    String FILE_SCHEME = "file";
+    String FILE_PROTOCOL = FILE_SCHEME + "://";
+
+    String LOCAL_FILES_CONFIG = "kestra.local-files";
+    String ALLOWED_PATHS_CONFIG = LocalPath.LOCAL_FILES_CONFIG + ".allowed-paths";
+    String ENABLE_FILE_FUNCTIONS_CONFIG = LocalPath.LOCAL_FILES_CONFIG + ".enable-file-functions";
+    String ENABLE_PREVIEW_CONFIG = LocalPath.LOCAL_FILES_CONFIG + ".enable-preview";
+
+
+    /**
+     * Get an InputStream of a local file denoted by this URI.
+     *
+     * @param uri a file URI
+     * @throws SecurityException if the file is not allowed globally or specifically for this plugin.
+     */
+    InputStream get(URI uri) throws IOException;
+
+    /**
+     * Return true if the local file denoted by this URI exists.
+     *
+     * @param uri a file URI
+     * @throws SecurityException if the file is not allowed globally or specifically for this plugin.
+     */
+    boolean exists(URI uri) throws IOException;
+
+    /**
+     * Get a local file attributes.
+     *
+     * @param uri a file URI
+     * @throws SecurityException if the file is not allowed globally or specifically for this plugin.
+     */
+    BasicFileAttributes getAttributes(URI uri) throws IOException;
+}

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

@@ -0,0 +1,134 @@
+package io.kestra.core.runners;
+
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collections;
+import java.util.List;
+
+@Singleton
+public class LocalPathFactory {
+    private final List<String> globalAllowedPaths;
+
+    @Inject
+    public LocalPathFactory(@Value("${" + LocalPath.ALLOWED_PATHS_CONFIG + ":}") List<String> globalAllowedPaths) {
+        this.globalAllowedPaths = globalAllowedPaths;
+    }
+
+    /**
+     * Create a LocalPath based on a RunContext, this is the preferred way as it would be allowed to check
+     * working directory and plugin configuration.
+     * If no RunContext is available {@link #createLocalPath()} can be used instead but this LocalPath would only be able to check
+     * paths globally allowed inside the Kestra configuration.
+     */
+    public LocalPath createLocalPath(RunContext runContext) {
+        return new RunContextLocalPath(globalAllowedPaths, runContext);
+    }
+
+    /**
+     * Create a LocalPath.
+     * If a RunContext is available, this is preferable to use {@link #createLocalPath(RunContext)} as it would be possible to
+     * check for paths inside the working directory or allowed inside the plugin configuration.
+     */
+    public LocalPath createLocalPath() {
+        return new DefaultLocalPath(globalAllowedPaths);
+    }
+
+    abstract static class AbstractLocalPath implements LocalPath {
+        @Override
+        public InputStream get(URI uri) throws IOException {
+            if (!LocalPath.FILE_SCHEME.equals(uri.getScheme())) {
+                throw new IllegalArgumentException("The uri '" + uri + "' is not a valid file URI.");
+            }
+
+            Path path = checkPath(uri);
+            return new FileInputStream(path.toFile());
+        }
+
+        @Override
+        public boolean exists(URI uri) throws IOException {
+            if (!LocalPath.FILE_SCHEME.equals(uri.getScheme())) {
+                throw new IllegalArgumentException("The uri '" + uri + "' is not a valid file URI.");
+            }
+
+            Path path = checkPath(uri);
+            return Files.exists(path);
+        }
+
+        @Override
+        public BasicFileAttributes getAttributes(URI uri) throws IOException {
+            if (!LocalPath.FILE_SCHEME.equals(uri.getScheme())) {
+                throw new IllegalArgumentException("The uri '" + uri + "' is not a valid file URI.");
+            }
+
+            Path path = checkPath(uri);
+            return Files.readAttributes(path, BasicFileAttributes.class);
+        }
+
+        /**
+         * Check the URI then return it as a Path.
+         * Based on the available context, implementors should:
+         * - check if the file is inside the working directory
+         * - check globally allowed paths
+         * - check if plugin allowed paths
+         */
+        protected abstract Path checkPath(URI uri) throws IOException;
+    }
+
+    static class RunContextLocalPath extends AbstractLocalPath {
+        private final List<String> globalAllowedPaths;
+        private final RunContext runContext;
+
+        RunContextLocalPath(List<String> globalAllowedPaths, RunContext runContext) {
+            this.globalAllowedPaths = globalAllowedPaths;
+            this.runContext = runContext;
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        protected Path checkPath(URI uri) throws IOException {
+            Path workingDirectory = runContext.workingDir().path(true);
+            Path path = Path.of(uri).toRealPath(); // toRealPath() will protect about path traversal issues
+            // We allow working directory or globally allowed path
+            if (!path.startsWith(workingDirectory) && globalAllowedPaths.stream().noneMatch(path::startsWith)) {
+                // if not globally allowed, we check if it's allowed for this specific plugin
+                List<String> pluginAllowedPaths = (List<String>) runContext.pluginConfiguration("allowed-paths").orElse(Collections.emptyList());
+                if (pluginAllowedPaths.stream().noneMatch(path::startsWith)) {
+                    throw new SecurityException("The path " + path + " is not authorized. " +
+                        "Only files inside the working directory are allowed by default, other path must be allowed either globally inside the Kestra configuration using the `" + LocalPath.ALLOWED_PATHS_CONFIG + "` property, " +
+                        "or by plugin using the `allowed-paths` plugin configuration.");
+                }
+            }
+
+            return path;
+        }
+    }
+
+    static class DefaultLocalPath extends AbstractLocalPath {
+        private final List<String> globalAllowedPaths;
+
+        DefaultLocalPath(List<String> globalAllowedPaths) {
+            this.globalAllowedPaths = globalAllowedPaths;
+        }
+
+        @Override
+        protected Path checkPath(URI uri) throws IOException {
+            Path path = Path.of(uri).toRealPath(); // toRealPath() will protect about path traversal issues
+            // we only allow globally allowed as we don't have a run context to get the working directory nor the plugin configuration
+            if (globalAllowedPaths.stream().noneMatch(path::startsWith)) {
+                throw new SecurityException("The path " + path + " is not authorized. " +
+                    "Path must be allowed either globally inside the Kestra configuration using the `" + LocalPath.ALLOWED_PATHS_CONFIG + "` property.");
+            }
+
+            return path;
+        }
+    }
+}

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

@@ -199,7 +199,10 @@ public class LocalWorkingDir implements WorkingDir {
 
         if (Files.exists(newFilePath)) {
             switch (comportment) {
-                case OVERWRITE -> copyFile(inputStream, newFilePath);
+                case OVERWRITE -> {
+                    log.info("File {} already exist. It will be overwritten", newFilePath);
+                    copyFile(inputStream, newFilePath);
+                }
                 case FAIL -> throw new FileAlreadyExistsException("File " + newFilePath + " already exist");
                 case WARN -> log.warn("File {} already exist. It will be ignore", newFilePath);
                 case IGNORE -> {}

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

@@ -13,7 +13,6 @@ import org.slf4j.Logger;
 
 import java.net.URI;
 import java.security.GeneralSecurityException;
-import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -180,6 +179,11 @@ public abstract class RunContext {
         return new StateStore(this, true);
     }
 
+    /**
+     * Get access to local paths of the host machine.
+     */
+    public abstract LocalPath localPath();
+
     public record FlowInfo(String tenantId, String namespace, String id, Integer revision) {
     }
 

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

@@ -108,6 +108,10 @@ 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);
+    }
+
+    public RunContext of(FlowInterface flow, Task task, Execution execution, TaskRun taskRun, boolean decryptVariables, VariableRenderer variableRenderer) {
         RunContextLogger runContextLogger = runContextLoggerFactory.create(taskRun, task, execution.getKind());
 
         return newBuilder()
@@ -127,6 +131,7 @@ public class RunContextFactory {
             .withKvStoreService(kvStoreService)
             .withSecretInputs(secretInputsFromFlow(flow))
             .withTask(task)
+            .withVariableRenderer(variableRenderer)
             .build();
     }
 

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

@@ -28,6 +28,7 @@ import java.util.function.Consumer;
  */
 public final class RunVariables {
     public static final String SECRET_CONSUMER_VARIABLE_NAME = "addSecretConsumer";
+    public static final String FIXTURE_FILES_KEY = "io.kestra.datatype:test_fixtures_files";
 
     /**
      * Creates an immutable map representation of the given {@link Task}.
@@ -214,7 +215,7 @@ public final class RunVariables {
 
                 executionMap.put("id", execution.getId());
 
-                if (execution.getState() != null) { // can occurs in tests
+                if (execution.getState() != null) { // can occur in tests
                     executionMap.put("state", execution.getState().getCurrent());
                 }
 
@@ -224,6 +225,10 @@ public final class RunVariables {
                 Optional.ofNullable(execution.getOriginalId())
                     .ifPresent(originalId -> executionMap.put("originalId", originalId));
 
+                if (execution.getOutputs() != null) {
+                    executionMap.put("outputs", execution.getOutputs());
+                }
+
                 builder.put("execution", executionMap.build());
 
                 if (execution.getTaskRunList() != null) {
@@ -299,7 +304,7 @@ public final class RunVariables {
 
                     // temporal hack to add back the `schedule`variables
                     // will be removed in 2.0
-                    if (trigger.getType().equals(Schedule.class.getName())) {
+                    if (Schedule.class.getName().equals(execution.getTrigger().getType())) {
                         // add back its variables inside the `schedule` variables
                         builder.put("schedule", execution.getTrigger().getVariables());
                     }
@@ -321,13 +326,18 @@ public final class RunVariables {
             }
 
             // variables
-            if (execution != null &&  execution.getVariables() != null) {
-                builder.put("vars", execution.getVariables());
-            }
-            else if (execution == null && flow != null && flow.getVariables() != null) {
-                // flow variables are added to the execution variables at execution creation time so they must only be added if the execution is null
-                builder.put("vars", flow.getVariables());
-            }
+            Optional.ofNullable(execution)
+                .map(Execution::getVariables)
+                .or(() -> Optional.ofNullable(flow).map(FlowInterface::getVariables))
+                .map(HashMap::new)
+                .ifPresent(variables -> {
+                    Object fixtureFiles = variables.remove(FIXTURE_FILES_KEY);
+                    builder.put("vars", ImmutableMap.copyOf(variables));
+
+                    if (fixtureFiles != null) {
+                        builder.put("files", fixtureFiles);
+                    }
+                });
 
             // Kestra configuration
             if (kestraConfiguration != null) {

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

@@ -2,22 +2,27 @@ 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.AbstractExtension;
+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 {
@@ -29,18 +34,26 @@ public class VariableRenderer {
 
     @Inject
     public VariableRenderer(ApplicationContext applicationContext, @Nullable VariableConfiguration variableConfiguration) {
+        this(applicationContext, variableConfiguration, Collections.emptyList());
+    }
+
+    public VariableRenderer(ApplicationContext applicationContext, @Nullable VariableConfiguration variableConfiguration, List<String> functionsToMask) {
         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);
 
-        applicationContext.getBeansOfType(AbstractExtension.class)
-            .forEach(pebbleBuilder::extension);
+        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()));
@@ -49,8 +62,63 @@ public class VariableRenderer {
         this.pebbleEngine = pebbleBuilder.build();
     }
 
-    public static IllegalVariableEvaluationException properPebbleException(PebbleException e) {
-        if (e instanceof AttributeNotFoundException current) {
+    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);
+            }
+        );
+    }
+
+    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(
                 "Unable to find `" + current.getAttributeName() +
                     "` used in the expression `" + current.getFileName() +
@@ -58,7 +126,7 @@ public class VariableRenderer {
             );
         }
 
-        return new IllegalVariableEvaluationException(e);
+        return new IllegalVariableEvaluationException(initialExtension);
     }
 
     public String render(String inline, Map<String, Object> variables) throws IllegalVariableEvaluationException {
@@ -133,10 +201,10 @@ public class VariableRenderer {
     /**
      * This method can be used in fallback for rendering an input string.
      *
-     * @param e         The exception that was throw by the default variable renderer.
-     * @param inline    The expression to be rendered.
-     * @param variables The context variables.
-     * @return          The rendered string.
+     * @param e The exception that was throw by the default variable renderer.
+     * @param inline           The expression to be rendered.
+     * @param variables        The context variables.
+     * @return The rendered string.
      */
     protected String alternativeRender(Exception e, String inline, Map<String, Object> variables) throws IllegalVariableEvaluationException {
         return null;

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

@@ -1,5 +1,6 @@
 package io.kestra.core.runners;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Throwables;
@@ -10,6 +11,7 @@ import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.tasks.Output;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.*;
@@ -18,6 +20,7 @@ import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.server.*;
 import io.kestra.core.services.LabelService;
 import io.kestra.core.services.LogService;
+import io.kestra.core.services.MaintenanceService;
 import io.kestra.core.services.VariablesService;
 import io.kestra.core.services.WorkerGroupService;
 import io.kestra.core.storages.StorageContext;
@@ -42,7 +45,14 @@ import org.apache.commons.lang3.time.StopWatch;
 import org.slf4j.Logger;
 import org.slf4j.event.Level;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
@@ -55,6 +65,9 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
 
 import static io.kestra.core.models.flows.State.Type.*;
 import static io.kestra.core.server.Service.ServiceState.TERMINATED_FORCED;
@@ -94,6 +107,10 @@ public class Worker implements Service, Runnable, AutoCloseable {
     @Named(QueueFactoryInterface.TRIGGER_NAMED)
     private QueueInterface<Trigger> triggerQueue;
 
+    @Inject
+    @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
+    private QueueInterface<LogEntry> logQueue;
+
     @Inject
     @Named(QueueFactoryInterface.CLUSTER_EVENT_NAMED)
     private Optional<QueueInterface<ClusterEvent>> clusterEventQueue;
@@ -158,6 +175,9 @@ public class Worker implements Service, Runnable, AutoCloseable {
     private TracerFactory tracerFactory;
     private Tracer tracer;
 
+    @Inject
+    private MaintenanceService maintenanceService;
+
     /**
      * Creates a new {@link Worker} instance.
      *
@@ -285,8 +305,12 @@ public class Worker implements Service, Runnable, AutoCloseable {
         ));
 
         this.clusterEventQueue.ifPresent(clusterEventQueueInterface -> this.receiveCancellations.addFirst(clusterEventQueueInterface.receive(this::clusterEventQueue)));
+        if (this.maintenanceService.isInMaintenanceMode()) {
+            enterMaintenance();
+        } else {
+            setState(ServiceState.RUNNING);
+        }
 
-        setState(ServiceState.RUNNING);
         if (workerGroupKey != null) {
             log.info("Worker started with {} thread(s) in group '{}'", numThreads, workerGroupKey);
         }
@@ -304,21 +328,25 @@ public class Worker implements Service, Runnable, AutoCloseable {
         ClusterEvent clusterEvent = either.getLeft();
         log.info("Cluster event received: {}", clusterEvent);
         switch (clusterEvent.eventType()) {
-            case MAINTENANCE_ENTER -> {
-                this.executionKilledQueue.pause();
-                this.workerJobQueue.pause();
-
-                this.setState(ServiceState.MAINTENANCE);
-            }
-            case MAINTENANCE_EXIT -> {
-                this.executionKilledQueue.resume();
-                this.workerJobQueue.resume();
-
-                this.setState(ServiceState.RUNNING);
-            }
+            case MAINTENANCE_ENTER -> enterMaintenance();
+            case MAINTENANCE_EXIT -> exitMaintenance();
         }
     }
 
+    private void enterMaintenance() {
+        this.executionKilledQueue.pause();
+        this.workerJobQueue.pause();
+
+        this.setState(ServiceState.MAINTENANCE);
+    }
+
+    private void exitMaintenance() {
+        this.executionKilledQueue.resume();
+        this.workerJobQueue.resume();
+
+        this.setState(ServiceState.RUNNING);
+    }
+
     private void setState(final ServiceState state) {
         this.state.set(state);
         Map<String, Object> properties = new HashMap<>();
@@ -338,7 +366,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
                 } else if ("trigger".equals(type)) {
                     // try to deserialize the triggerContext to fail it
                     var triggerContext = MAPPER.treeToValue(json.get("triggerContext"), TriggerContext.class);
-                    var workerTriggerResult = WorkerTriggerResult.builder().triggerContext(triggerContext).success(false).execution(Optional.empty()).build();
+                    var workerTriggerResult = WorkerTriggerResult.builder().triggerContext(triggerContext).execution(Optional.empty()).build();
                     this.workerTriggerResultQueue.emit(workerTriggerResult);
                 }
             } catch (IOException | QueueException e) {
@@ -477,11 +505,23 @@ public class Worker implements Service, Runnable, AutoCloseable {
 
         logError(workerTrigger, e);
         try {
+            Execution execution = workerTrigger.getTrigger().isFailOnTriggerError() ? TriggerService.generateExecution(workerTrigger.getTrigger(), workerTrigger.getConditionContext(), workerTrigger.getTriggerContext(), (Output) null)
+                .withState(FAILED) : null;
+            if (execution != null) {
+                RunContextLogger.logEntries(Execution.loggingEventFromException(e), LogEntry.of(execution))
+                    .forEach(log -> {
+                        try {
+                            logQueue.emitAsync(log);
+                        } catch (QueueException ex) {
+                            // fail silently
+                        }
+                    });
+            }
             this.workerTriggerResultQueue.emit(
                 WorkerTriggerResult.builder()
-                    .success(false)
                     .triggerContext(workerTrigger.getTriggerContext())
                     .trigger(workerTrigger.getTrigger())
+                    .execution(Optional.ofNullable(execution))
                     .build()
             );
         } catch (QueueException ex) {
@@ -518,7 +558,6 @@ public class Worker implements Service, Runnable, AutoCloseable {
         try {
             this.workerTriggerResultQueue.emit(
                 WorkerTriggerResult.builder()
-                    .success(false)
                     .execution(Optional.of(execution))
                     .triggerContext(workerTrigger.getTriggerContext())
                     .trigger(workerTrigger.getTrigger())
@@ -629,7 +668,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
                 ));
         }
 
-        if (killedExecution.contains(workerTask.getTaskRun().getExecutionId())) {
+        if (! Boolean.TRUE.equals(workerTask.getTaskRun().getForceExecution()) && killedExecution.contains(workerTask.getTaskRun().getExecutionId())) {
             WorkerTaskResult workerTaskResult = new WorkerTaskResult(workerTask.getTaskRun().withState(KILLED));
             try {
                 this.workerTaskResultQueue.emit(workerTaskResult);
@@ -654,9 +693,44 @@ public class Worker implements Service, Runnable, AutoCloseable {
 
         workerTask = workerTask.withTaskRun(workerTask.getTaskRun().withState(RUNNING));
 
+        DefaultRunContext runContext = runContextInitializer.forWorker((DefaultRunContext) workerTask.getRunContext(), workerTask);
+        Optional<String> hash = Optional.empty();
+        if (workerTask.getTask().getTaskCache() != null && workerTask.getTask().getTaskCache().getEnabled()) {
+            runContext.logger().debug("Task output caching is enabled for task '{}''", workerTask.getTask().getId());
+            hash = hashTask(runContext, workerTask.getTask());
+            if (hash.isPresent()) {
+                try {
+                    Optional<InputStream> cacheFile = runContext.storage().getCacheFile(hash.get(), workerTask.getTaskRun().getValue(), workerTask.getTask().getTaskCache().getTtl());
+                    if (cacheFile.isPresent()) {
+                        runContext.logger().info("Skipping task execution for task '{}' as there is an existing cache entry for it", workerTask.getTask().getId());
+                        try (ZipInputStream archive = new ZipInputStream(cacheFile.get())) {
+                            if (archive.getNextEntry() != null) {
+                                byte[] cache = archive.readAllBytes();
+                                Map<String, Object> outputMap = JacksonMapper.ofIon().readValue(cache, JacksonMapper.MAP_TYPE_REFERENCE);
+                                Variables variables = variablesService.of(StorageContext.forTask(workerTask.getTaskRun()), outputMap);
+
+                                TaskRunAttempt attempt = TaskRunAttempt.builder()
+                                    .state(new io.kestra.core.models.flows.State().withState(SUCCESS))
+                                    .workerId(this.id)
+                                    .build();
+                                List<TaskRunAttempt> attempts = this.addAttempt(workerTask, attempt);
+                                TaskRun taskRun = workerTask.getTaskRun().withAttempts(attempts).withOutputs(variables).withState(SUCCESS);
+                                WorkerTaskResult workerTaskResult = new WorkerTaskResult(taskRun);
+                                this.workerTaskResultQueue.emit(workerTaskResult);
+                                return workerTaskResult;
+                            }
+                        }
+                    }
+                } catch (IOException | RuntimeException | QueueException e) {
+                    // in case of any exception, log an error and continue
+                    runContext.logger().error("Unexpected exception while loading the cache for task '{}', the task will be executed instead.", workerTask.getTask().getId(), e);
+                }
+            }
+        }
+
         try {
             // run
-            workerTask = this.runAttempt(workerTask);
+            workerTask = this.runAttempt(runContext, workerTask);
 
             // get last state
             TaskRunAttempt lastAttempt = workerTask.getTaskRun().lastAttempt();
@@ -691,6 +765,28 @@ public class Worker implements Service, Runnable, AutoCloseable {
 
             WorkerTaskResult workerTaskResult = new WorkerTaskResult(workerTask.getTaskRun(), dynamicTaskRuns);
             this.workerTaskResultQueue.emit(workerTaskResult);
+
+            // upload the cache file, hash may not be present if we didn't succeed in computing it
+            if (workerTask.getTask().getTaskCache() != null && workerTask.getTask().getTaskCache().getEnabled() && hash.isPresent() &&
+                (state == State.Type.SUCCESS || state == State.Type.WARNING)) {
+                runContext.logger().info("Uploading a cache entry for task '{}'", workerTask.getTask().getId());
+
+                try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
+                     ZipOutputStream archive = new ZipOutputStream(bos)) {
+                    var zipEntry = new ZipEntry("outputs.ion");
+                    archive.putNextEntry(zipEntry);
+                    archive.write(JacksonMapper.ofIon().writeValueAsBytes(workerTask.getTaskRun().getOutputs()));
+                    archive.closeEntry();
+                    archive.finish();
+                    Path archiveFile = runContext.workingDir().createTempFile( ".zip");
+                    Files.write(archiveFile, bos.toByteArray());
+                    URI uri = runContext.storage().putCacheFile(archiveFile.toFile(), hash.get(), workerTask.getTaskRun().getValue());
+                    runContext.logger().debug("Caching entry uploaded in URI {}", uri);
+                } catch (IOException | RuntimeException e) {
+                    // in case of any exception, log an error and continue
+                    runContext.logger().error("Unexpected exception while uploading the cache entry for task '{}', the task not be cached.", workerTask.getTask().getId(), e);
+                }
+            }
             return workerTaskResult;
         } catch (QueueException e) {
             // If there is a QueueException it can either be caused by the message limit or another queue issue.
@@ -719,6 +815,22 @@ public class Worker implements Service, Runnable, AutoCloseable {
         }
     }
 
+    private Optional<String> hashTask(RunContext runContext, Task task) {
+        try {
+            var map = JacksonMapper.toMap(task);
+            var rMap = runContext.render(map);
+            var json = JacksonMapper.ofJson().writeValueAsBytes(rMap);
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            digest.update(json);
+            byte[] bytes = digest.digest();
+            return Optional.of(HexFormat.of().formatHex(bytes));
+        } catch (RuntimeException | IllegalVariableEvaluationException | JsonProcessingException |
+                 NoSuchAlgorithmException e) {
+            runContext.logger().error("Unable to create the cache key for the task '{}'", task.getId(), e);
+            return Optional.empty();
+        }
+    }
+
     private List<TaskRun> dynamicWorkerResults(List<WorkerTaskResult> dynamicWorkerResults) {
         return dynamicWorkerResults
             .stream()
@@ -774,9 +886,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
         }
     }
 
-    private WorkerTask runAttempt(final WorkerTask workerTask) throws QueueException {
-        DefaultRunContext runContext = runContextInitializer.forWorker((DefaultRunContext) workerTask.getRunContext(), workerTask);
-
+    private WorkerTask runAttempt(RunContext runContext, final WorkerTask workerTask) throws QueueException {
         Logger logger = runContext.logger();
 
         if (!(workerTask.getTask() instanceof RunnableTask<?> task)) {
@@ -1038,18 +1148,16 @@ public class Worker implements Service, Runnable, AutoCloseable {
         }
     }
 
+    /**
+     * This method should only be used on tests.
+     * It shut down the worker without waiting for tasks to end,
+     * and without closing the queue, so tests can launch and shutdown a worker manually without closing the queue.
+     */
     @VisibleForTesting
     public void shutdown() {
         // initiate shutdown
         shutdown.compareAndSet(false, true);
 
-        try {
-            // close the WorkerJob queue to stop receiving new JobTask execution.
-            workerJobQueue.close();
-        } catch (IOException e) {
-            log.error("Failed to close the WorkerJobQueue");
-        }
-
         // close all queues and shutdown now
         this.receiveCancellations.forEach(Runnable::run);
         this.executorService.shutdownNow();

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

@@ -25,9 +25,6 @@ public class WorkerTriggerResult implements HasUID {
     @NotNull
     AbstractTrigger trigger;
 
-    @Builder.Default
-    Boolean success = true;
-
     /**
      * {@inheritDoc}
      */

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

@@ -17,9 +17,11 @@ public abstract class AbstractDate {
 
     private static final Map<String, DateTimeFormatter> FORMATTERS = ImmutableMap.<String, DateTimeFormatter>builder()
         .put("iso", DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSSXXX"))
+        .put("iso_milli", DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX"))
         .put("iso_sec", DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssXXX"))
         .put("sql", DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS"))
-        .put("sql_seq", DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
+        .put("sql_milli", DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"))
+        .put("sql_sec", DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
         .put("iso_date_time", DateTimeFormatter.ISO_DATE_TIME)
         .put("iso_date", DateTimeFormatter.ISO_DATE)
         .put("iso_time", DateTimeFormatter.ISO_TIME)