Merge branch 'develop' into fix/failin_execution_test

# Conflicts:
#	webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java

.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"
@@ -20,6 +20,15 @@ on:
         required: false
         type: string
         default: "LATEST"
+      force-download-artifact:
+        description: 'Force download artifact'
+        required: false
+        type: string
+        default: "true"
+        options:
+          - "true"
+          - "false"
+
 env:
   PLUGIN_VERSION: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
 jobs:
@@ -38,9 +47,18 @@ jobs:
         id: plugins
         with:
           plugin-version: ${{ env.PLUGIN_VERSION }}
+
+  # ********************************************************************************************************************
+  # Build
+  # ********************************************************************************************************************
+  build-artifacts:
+    name: Build Artifacts
+    if: ${{ github.event.inputs.force-download-artifact == 'true' }}
+    uses: ./.github/workflows/workflow-build-artifacts.yml
+
   docker:
     name: Publish Docker
-    needs: [ plugins ]
+    needs: [ plugins, build-artifacts ]
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -69,18 +87,31 @@ 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
-      # Download release
-      - name: Download release
+
+      # [workflow_dispatch]
+      # Download executable from GitHub Release
+      - name: Artifacts - Download release (workflow_dispatch)
+        id: download-github-release
+        if: github.event_name == 'workflow_dispatch' && github.event.inputs.force-download-artifact == 'false'
         uses: robinraju/release-downloader@v1.12
         with:
           tag: ${{steps.vars.outputs.tag}}
           fileName: 'kestra-*'
           out-file-path: build/executable
 
+      # [workflow_call]
+      # Download executable from artifact
+      - name: Artifacts - Download executable
+        if: github.event_name != 'workflow_dispatch' || steps.download-github-release.outcome == 'skipped'
+        uses: actions/download-artifact@v4
+        with:
+          name: exe
+          path: build/executable
+
       - name: Copy exe to image
         run: |
           cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra

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

@@ -43,7 +43,8 @@ jobs:
       SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
       SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
       SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
-
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
   end:
     runs-on: ubuntu-latest
     needs:

.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

@@ -1,23 +1,7 @@
 name: Build Artifacts
 
 on:
-  workflow_call:
-    inputs:
-      plugin-version:
-        description: "Kestra version"
-        default: 'LATEST'
-        required: true
-        type: string
-    outputs:
-      docker-tag:
-        value: ${{ jobs.build.outputs.docker-tag }}
-        description: "The Docker image Tag for Kestra"
-      docker-artifact-name:
-        value: ${{ jobs.build.outputs.docker-artifact-name }}
-        description: "The GitHub artifact containing the Kestra docker image name."
-      plugins:
-        value: ${{ jobs.build.outputs.plugins }}
-        description: "The Kestra plugins list used for the build."
+  workflow_call: {}
 
 jobs:
   build:
@@ -68,7 +52,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
@@ -82,55 +66,6 @@ jobs:
         run: |
           cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
 
-      # Docker Tag
-      - name: Setup - Docker vars
-        id: vars
-        shell: bash
-        run: |
-          TAG=${GITHUB_REF#refs/*/}
-          if [[ $TAG = "master" ]]
-          then
-            TAG="latest";
-          elif [[ $TAG = "develop" ]]
-          then
-            TAG="develop";
-          elif [[ $TAG = v* ]]
-          then
-            TAG="${TAG}";
-          else
-            TAG="build-${{ github.run_id }}";
-          fi
-          echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          echo "artifact=docker-kestra-${TAG}" >> $GITHUB_OUTPUT
-
-      # Docker setup
-      - name: Docker - Setup QEMU
-        uses: docker/setup-qemu-action@v3
-
-      - name: Docker - Fix Qemu
-        shell: bash
-        run: |
-          docker run --rm --privileged multiarch/qemu-user-static --reset -p yes -c yes
-
-      - name: Docker - Setup Buildx
-        uses: docker/setup-buildx-action@v3
-
-      # Docker Build
-      - name: Docker - Build & export image
-        uses: docker/build-push-action@v6
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        with:
-          context: .
-          push: false
-          file: Dockerfile
-          tags: |
-            kestra/kestra:${{ steps.vars.outputs.tag }}
-          build-args: |
-            KESTRA_PLUGINS=${{ steps.plugins.outputs.plugins }}
-            APT_PACKAGES=${{ env.DOCKER_APT_PACKAGES }}
-            PYTHON_LIBRARIES=${{ env.DOCKER_PYTHON_LIBRARIES }}
-          outputs: type=docker,dest=/tmp/${{ steps.vars.outputs.artifact }}.tar
-
       # Upload artifacts
       - name: Artifacts - Upload JAR
         uses: actions/upload-artifact@v4
@@ -143,10 +78,3 @@ jobs:
         with:
           name: exe
           path: build/executable/
-
-      - name: Artifacts - Upload Docker
-        uses: actions/upload-artifact@v4
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        with:
-          name: ${{ steps.vars.outputs.artifact }}
-          path: /tmp/${{ steps.vars.outputs.artifact }}.tar

.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

@@ -1,14 +1,17 @@
 name: Github - Release
 
 on:
+  workflow_dispatch:
   workflow_call:
     secrets:
       GH_PERSONAL_TOKEN:
         description: "The Github personal token."
         required: true
-  push:
-    tags:
-      - '*'
+      SLACK_RELEASES_WEBHOOK_URL:
+        description: "The Slack webhook URL."
+        required: true
+
+
 
 jobs:
   publish:
@@ -41,12 +44,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

@@ -41,8 +41,6 @@ jobs:
     name: Build Artifacts
     if: ${{ github.event.inputs.force-download-artifact == 'true' }}
     uses: ./.github/workflows/workflow-build-artifacts.yml
-    with:
-      plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
   # ********************************************************************************************************************
   # Docker
   # ********************************************************************************************************************
@@ -112,12 +110,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

.github/workflows/workflow-release.yml

@@ -42,12 +42,16 @@ on:
       SONATYPE_GPG_FILE:
         description: "The Sonatype GPG file."
         required: true
+      GH_PERSONAL_TOKEN:
+        description: "GH personnal Token."
+        required: true
+      SLACK_RELEASES_WEBHOOK_URL:
+        description: "Slack webhook for releases channel."
+        required: true
 jobs:
   build-artifacts:
     name: Build - Artifacts
     uses: ./.github/workflows/workflow-build-artifacts.yml
-    with:
-      plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
 
   Docker:
     name: Publish Docker
@@ -77,4 +81,5 @@ jobs:
     if: startsWith(github.ref, 'refs/tags/v')
     uses: ./.github/workflows/workflow-github-release.yml
     secrets:
-      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}

.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
@@ -24,11 +26,13 @@
 #plugin-debezium:io.kestra.plugin:plugin-debezium-oracle:LATEST
 #plugin-debezium:io.kestra.plugin:plugin-debezium-postgres:LATEST
 #plugin-debezium:io.kestra.plugin:plugin-debezium-sqlserver:LATEST
+#plugin-deepseek:io.kestra.plugin:plugin-deepseek:LATEST
 #plugin-docker:io.kestra.plugin:plugin-docker:LATEST
 #plugin-elasticsearch:io.kestra.plugin:plugin-elasticsearch:LATEST
 #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,31 +67,38 @@
 #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
+#plugin-scripts:io.kestra.plugin:plugin-script-bun:LATEST
+#plugin-scripts:io.kestra.plugin:plugin-script-deno:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-go:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-groovy:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-jbang:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-julia:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-jython:LATEST
+#plugin-scripts:io.kestra.plugin:plugin-script-lua:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-nashorn:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-node:LATEST
+#plugin-scripts:io.kestra.plugin:plugin-script-perl:LATEST
+#plugin-scripts:io.kestra.plugin:plugin-script-php:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-powershell:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-python:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-r: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'
@@ -221,14 +225,14 @@ subprojects {
         }
 
         testlogger {
-            theme 'mocha-parallel'
-            showExceptions true
-            showFullStackTraces true
-            showCauses true
-            slowThreshold 2000
-            showStandardStreams true
-            showPassedStandardStreams false
-            showSkippedStandardStreams true
+            theme = 'mocha-parallel'
+            showExceptions = true
+            showFullStackTraces = true
+            showCauses = true
+            slowThreshold = 2000
+            showStandardStreams = true
+            showPassedStandardStreams = false
+            showSkippedStandardStreams = true
         }
     }
 }
@@ -406,7 +410,7 @@ jar {
 shadowJar {
     archiveClassifier.set(null)
     mergeServiceFiles()
-    zip64 true
+    zip64 = true
 }
 
 distZip.dependsOn shadowJar
@@ -414,6 +418,7 @@ distTar.dependsOn shadowJar
 startScripts.dependsOn shadowJar
 startShadowScripts.dependsOn jar
 shadowJar.dependsOn 'ui:assembleFrontend'
+shadowJar.dependsOn jar
 
 /**********************************************************************************************************************\
  * Executable Jar
@@ -422,8 +427,8 @@ def executableDir = layout.buildDirectory.dir("executable")
 def executable = layout.buildDirectory.file("executable/${project.name}-${project.version}").get().asFile
 
 tasks.register('writeExecutableJar') {
-    group "build"
-    description "Write an executable jar from shadow jar"
+    group = "build"
+    description = "Write an executable jar from shadow jar"
     dependsOn = [shadowJar]
 
     final shadowJarFile =  tasks.shadowJar.outputs.files.singleFile
@@ -449,8 +454,8 @@ tasks.register('writeExecutableJar') {
 }
 
 tasks.register('executableJar', Zip) {
-    group "build"
-    description "Zip the executable jar"
+    group = "build"
+    description = "Zip the executable jar"
     dependsOn = [writeExecutableJar]
 
     archiveFileName = "${project.name}-${project.version}.zip"
@@ -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/FileChangedEventListener.java

@@ -162,7 +162,15 @@ public class FileChangedEventListener {
                                     }
 
                                 } catch (NoSuchFileException e) {
-                                    log.error("File not found: {}", entry, e);
+                                    log.warn("File not found: {}, deleting it", entry, e);
+                                    // the file might have been deleted while reading so if not found we try to delete the flow
+                                    flows.stream()
+                                        .filter(flow -> flow.getPath().equals(filePath.toString()))
+                                        .findFirst()
+                                        .ifPresent(flowWithPath -> {
+                                            flowFilesManager.deleteFlow(flowWithPath.getTenantId(), flowWithPath.getNamespace(), flowWithPath.getId());
+                                            this.flows.removeIf(fwp -> fwp.uidWithoutRevision().equals(flowWithPath.uidWithoutRevision()));
+                                        });
                                 } catch (IOException e) {
                                     log.error("Error reading file: {}", entry, e);
                                 }

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

@@ -1,11 +1,10 @@
 package io.kestra.core.models;
 
 import io.kestra.core.utils.MapUtils;
+import jakarta.annotation.Nullable;
 import jakarta.validation.constraints.NotNull;
 
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 
 public record Label(@NotNull String key, @NotNull String value) {
@@ -29,11 +28,36 @@ public record Label(@NotNull String key, @NotNull String value) {
      * @return the nested {@link Map}.
      */
     public static Map<String, Object> toNestedMap(List<Label> labels) {
-        Map<String, Object> asMap = labels.stream()
+        return MapUtils.flattenToNestedMap(toMap(labels));
+    }
+
+    /**
+     * Static helper method for converting a list of labels to a flat map.
+     * Key order is kept.
+     *
+     * @param labels The list of {@link Label} to be converted.
+     * @return the flat {@link Map}.
+     */
+    public static Map<String, String> toMap(@Nullable List<Label> labels) {
+        if (labels == null || labels.isEmpty()) return Collections.emptyMap();
+        return labels.stream()
             .filter(label -> label.value() != null && label.key() != null)
-            // using an accumulator in case labels with the same key exists: the first is kept
-            .collect(Collectors.toMap(Label::key, Label::value, (first, second) -> first));
-        return MapUtils.flattenToNestedMap(asMap);
+            // using an accumulator in case labels with the same key exists: the second is kept
+            .collect(Collectors.toMap(Label::key, Label::value, (first, second) -> second, LinkedHashMap::new));
+    }
+
+    /**
+     * Static helper method for deduplicating a list of labels by their key.
+     * Value of the last key occurrence is kept.
+     *
+     * @param labels The list of {@link Label} to be deduplicated.
+     * @return the deduplicated {@link List}.
+     */
+    public static List<Label> deduplicate(@Nullable List<Label> labels) {
+        if (labels == null || labels.isEmpty()) return Collections.emptyList();
+        return toMap(labels).entrySet().stream()
+            .map(entry -> new Label(entry.getKey(), entry.getValue()))
+            .collect(Collectors.toCollection(ArrayList::new));
     }
 
     /**

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

@@ -6,9 +6,9 @@ import com.fasterxml.jackson.annotation.JsonValue;
 import io.kestra.core.exceptions.InvalidQueryFiltersException;
 import io.kestra.core.models.dashboards.filters.*;
 import io.kestra.core.utils.Enums;
-import java.util.ArrayList;
 import lombok.Builder;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -49,42 +49,27 @@ public record QueryFilter(
         PREFIX
     }
 
+    @SuppressWarnings("unchecked")
     private List<Object> asValues(Object value) {
         return value instanceof String valueStr ? Arrays.asList(valueStr.split(",")) : (List<Object>) value;
     }
 
-    @SuppressWarnings("unchecked")
     public <T extends Enum<T>> AbstractFilter<T> toDashboardFilterBuilder(T field, Object value) {
-        switch (this.operation) {
-            case EQUALS:
-                return EqualTo.<T>builder().field(field).value(value).build();
-            case NOT_EQUALS:
-                return NotEqualTo.<T>builder().field(field).value(value).build();
-            case GREATER_THAN:
-                return GreaterThan.<T>builder().field(field).value(value).build();
-            case LESS_THAN:
-                return LessThan.<T>builder().field(field).value(value).build();
-            case GREATER_THAN_OR_EQUAL_TO:
-                return GreaterThanOrEqualTo.<T>builder().field(field).value(value).build();
-            case LESS_THAN_OR_EQUAL_TO:
-                return LessThanOrEqualTo.<T>builder().field(field).value(value).build();
-            case IN:
-                return In.<T>builder().field(field).values(asValues(value)).build();
-            case NOT_IN:
-                return NotIn.<T>builder().field(field).values(asValues(value)).build();
-            case STARTS_WITH:
-                return StartsWith.<T>builder().field(field).value(value.toString()).build();
-            case ENDS_WITH:
-                return EndsWith.<T>builder().field(field).value(value.toString()).build();
-            case CONTAINS:
-                return Contains.<T>builder().field(field).value(value.toString()).build();
-            case REGEX:
-                return Regex.<T>builder().field(field).value(value.toString()).build();
-            case PREFIX:
-                return Regex.<T>builder().field(field).value("^" + value.toString().replace(".", "\\.") + "(?:\\..+)?$").build();
-            default:
-                throw new IllegalArgumentException("Unsupported operation: " + this.operation);
-        }
+        return switch (this.operation) {
+            case EQUALS -> EqualTo.<T>builder().field(field).value(value).build();
+            case NOT_EQUALS -> NotEqualTo.<T>builder().field(field).value(value).build();
+            case GREATER_THAN -> GreaterThan.<T>builder().field(field).value(value).build();
+            case LESS_THAN -> LessThan.<T>builder().field(field).value(value).build();
+            case GREATER_THAN_OR_EQUAL_TO -> GreaterThanOrEqualTo.<T>builder().field(field).value(value).build();
+            case LESS_THAN_OR_EQUAL_TO -> LessThanOrEqualTo.<T>builder().field(field).value(value).build();
+            case IN -> In.<T>builder().field(field).values(asValues(value)).build();
+            case NOT_IN -> NotIn.<T>builder().field(field).values(asValues(value)).build();
+            case STARTS_WITH -> StartsWith.<T>builder().field(field).value(value.toString()).build();
+            case ENDS_WITH -> EndsWith.<T>builder().field(field).value(value.toString()).build();
+            case CONTAINS -> Contains.<T>builder().field(field).value(value.toString()).build();
+            case REGEX -> Regex.<T>builder().field(field).value(value.toString()).build();
+            case PREFIX -> Regex.<T>builder().field(field).value("^" + value.toString().replace(".", "\\.") + "(?:\\..+)?$").build();
+        };
     }
 
     public enum Field {

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;
@@ -24,6 +25,7 @@ import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import io.kestra.core.services.LabelService;
 import io.kestra.core.test.flow.TaskFixture;
 import io.kestra.core.utils.IdUtils;
+import io.kestra.core.utils.ListUtils;
 import io.kestra.core.utils.MapUtils;
 import io.swagger.v3.oas.annotations.Hidden;
 import jakarta.annotation.Nullable;
@@ -120,6 +122,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}.
      *
@@ -127,12 +132,12 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param labels The Flow labels.
      * @return a new {@link Execution}.
      */
-    public static Execution newExecution(final Flow flow, final List<Label> labels) {
+    public static Execution newExecution(final FlowInterface flow, final List<Label> labels) {
         return newExecution(flow, null, labels, Optional.empty());
     }
 
     public List<Label> getLabels() {
-        return Optional.ofNullable(this.labels).orElse(new ArrayList<>());
+        return ListUtils.emptyOnNull(this.labels);
     }
 
     /**
@@ -177,8 +182,22 @@ public class Execution implements DeletedInterface, TenantInterface {
     }
 
 
+    /**
+     * Customization of Lombok-generated builder.
+     */
     public static class ExecutionBuilder {
 
+        /**
+         * Enforce unique values of {@link Label} when using the builder.
+         *
+         * @param labels The labels.
+         * @return Deduplicated labels.
+         */
+        public ExecutionBuilder labels(List<Label> labels) {
+            this.labels = Label.deduplicate(labels);
+            return this;
+        }
+
         void prebuild() {
             this.originalId = this.id;
             this.metadata = ExecutionMetadata.builder()
@@ -221,12 +240,12 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
         );
     }
 
     public Execution withLabels(List<Label> labels) {
-
         return new Execution(
             this.tenantId,
             this.id,
@@ -236,7 +255,7 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.taskRunList,
             this.inputs,
             this.outputs,
-            labels,
+            Label.deduplicate(labels),
             this.variables,
             this.state,
             this.parentId,
@@ -247,7 +266,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
         );
     }
 
@@ -286,7 +306,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 +359,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.scheduleDate,
             this.traceParent,
             this.fixtures,
-            this.kind
+            this.kind,
+            this.breakpoints
         );
     }
 
@@ -366,7 +414,7 @@ public class Execution implements DeletedInterface, TenantInterface {
      *
      * @param resolvedTasks normal tasks
      * @param resolvedErrors errors tasks
-     * @param resolvedErrors finally tasks
+     * @param resolvedFinally finally tasks
      * @return the flow we need to follow
      */
     public List<ResolvedTask> findTaskDependingFlowState(
@@ -824,7 +872,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;
@@ -36,6 +38,8 @@ public abstract class AbstractFlow implements FlowInterface {
     @Min(value = 1)
     Integer revision;
 
+    String description;
+
     @Valid
     List<Input<?>> inputs;
 
@@ -60,6 +64,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;
@@ -64,12 +61,11 @@ public class Flow extends AbstractFlow implements HasUID {
             }
         });
 
-    String description;
-
     Map<String, Object> variables;
 
     @Valid
     @NotEmpty
+    @Schema(additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
     List<Task> tasks;
 
     @Valid
@@ -187,19 +183,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;
@@ -30,6 +31,8 @@ public interface FlowInterface extends FlowId, DeletedInterface, TenantInterface
 
     Pattern YAML_REVISION_MATCHER = Pattern.compile("(?m)^revision: \\d+\n?");
 
+    String getDescription();
+
     boolean isDisabled();
 
     boolean isDeleted();
@@ -42,6 +45,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

@@ -20,8 +20,8 @@ public class FileInput extends Input<URI> {
 
     private static final String DEFAULT_EXTENSION = ".upl";
 
-    @Builder.Default
-    public String extension = DEFAULT_EXTENSION;
+    @Deprecated(since = "0.24", forRemoval = true)
+    public String extension;
 
     @Override
     public void validate(URI input) throws ConstraintViolationException {
@@ -32,6 +32,7 @@ public class FileInput extends Input<URI> {
         String res = inputs.stream()
             .filter(in -> in instanceof FileInput)
             .filter(in -> in.getId().equals(fileName))
+            .filter(flowInput -> ((FileInput) flowInput).getExtension() != null)
             .map(flowInput -> ((FileInput) flowInput).getExtension())
             .findFirst()
             .orElse(FileInput.DEFAULT_EXTENSION);

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/queues/QueueFactoryInterface.java

@@ -28,6 +28,7 @@ public interface QueueFactoryInterface {
     String SUBFLOWEXECUTIONRESULT_NAMED = "subflowExecutionResultQueue";
     String CLUSTER_EVENT_NAMED = "clusterEventQueue";
     String SUBFLOWEXECUTIONEND_NAMED = "subflowExecutionEndQueue";
+    String EXECUTION_RUNNING_NAMED = "executionRunningQueue";
 
     QueueInterface<Execution> execution();
 
@@ -58,4 +59,6 @@ public interface QueueFactoryInterface {
     QueueInterface<SubflowExecutionResult> subflowExecutionResult();
 
     QueueInterface<SubflowExecutionEnd> subflowExecutionEnd();
+
+    QueueInterface<ExecutionRunning> executionRunning();
 }

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

@@ -27,7 +27,7 @@ public interface QueueInterface<T> extends Closeable, Pauseable {
     void delete(String consumerGroup, T message) throws QueueException;
 
     default Runnable receive(Consumer<Either<T, DeserializationException>> consumer) {
-        return receive((String) null, consumer);
+        return receive(null, consumer, false);
     }
 
     default Runnable receive(String consumerGroup, Consumer<Either<T, DeserializationException>> consumer) {

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

@@ -27,8 +27,6 @@ public class QueueService {
             return ((Executor) object).getExecution().getId();
         } else if (object.getClass() == MetricEntry.class) {
             return null;
-        } else if (object.getClass() == ExecutionRunning.class) {
-            return ((ExecutionRunning) object).getExecution().getId();
         } else if (object.getClass() == SubflowExecutionEnd.class) {
             return ((SubflowExecutionEnd) object).getParentExecutionId();
         } else {

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(
+        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> {
     /**
@@ -84,23 +82,14 @@ public interface LogRepositoryInterface extends SaveRepositoryInterface<LogEntry
     Flux<LogEntry> findAsync(
         @Nullable String tenantId,
         @Nullable String namespace,
+        @Nullable String flowId,
+        @Nullable String executionId,
         @Nullable Level minLevel,
         ZonedDateTime startDate
     );
 
     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 +98,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

@@ -1,5 +1,6 @@
 package io.kestra.core.runners;
 
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.utils.IdUtils;
 import jakarta.validation.constraints.NotNull;
@@ -11,7 +12,7 @@ import lombok.With;
 @Value
 @AllArgsConstructor
 @Builder
-public class ExecutionRunning {
+public class ExecutionRunning implements HasUID {
     String tenantId;
 
     @NotNull
@@ -26,9 +27,10 @@ public class ExecutionRunning {
     @With
     ConcurrencyState concurrencyState;
 
+    @Override
     public String uid() {
         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;
@@ -100,49 +102,39 @@ public class ExecutorService {
         return this.flowExecutorInterface;
     }
 
-    public Executor checkConcurrencyLimit(Executor executor, FlowInterface flow, Execution execution, long count) {
-        // if above the limit, handle concurrency limit based on its behavior
-        if (count >= flow.getConcurrency().getLimit()) {
+    public ExecutionRunning processExecutionRunning(FlowInterface flow, int runningCount, ExecutionRunning executionRunning) {
+        // if concurrency was removed, it can be null as we always get the latest flow definition
+        if (flow.getConcurrency() != null && runningCount >= flow.getConcurrency().getLimit()) {
             return switch (flow.getConcurrency().getBehavior()) {
                 case QUEUE -> {
-                    var newExecution = execution.withState(State.Type.QUEUED);
-
-                    ExecutionRunning executionRunning = ExecutionRunning.builder()
-                        .tenantId(flow.getTenantId())
-                        .namespace(flow.getNamespace())
-                        .flowId(flow.getId())
-                        .execution(newExecution)
-                        .concurrencyState(ExecutionRunning.ConcurrencyState.QUEUED)
-                        .build();
-
-                    // when max concurrency is reached, we throttle the execution and stop processing
                     logService.logExecution(
-                        newExecution,
+                        executionRunning.getExecution(),
                         Level.INFO,
-                        "Flow is queued due to concurrency limit exceeded, {} running(s)",
-                        count
+                        "Execution is queued due to concurrency limit exceeded, {} running(s)",
+                        runningCount
                     );
-                    // return the execution queued
-                    yield executor
-                        .withExecutionRunning(executionRunning)
-                        .withExecution(newExecution, "checkConcurrencyLimit");
+                    var newExecution = executionRunning.getExecution().withState(State.Type.QUEUED);
+                    metricRegistry.counter(MetricRegistry.METRIC_EXECUTOR_EXECUTION_QUEUED_COUNT, MetricRegistry.METRIC_EXECUTOR_EXECUTION_QUEUED_COUNT_DESCRIPTION, metricRegistry.tags(newExecution)).increment();
+                    yield executionRunning
+                        .withExecution(newExecution)
+                        .withConcurrencyState(ExecutionRunning.ConcurrencyState.QUEUED);
                 }
                 case CANCEL ->
-                    executor.withExecution(execution.withState(State.Type.CANCELLED), "checkConcurrencyLimit");
+                    executionRunning
+                        .withExecution(executionRunning.getExecution().withState(State.Type.CANCELLED))
+                        .withConcurrencyState(ExecutionRunning.ConcurrencyState.RUNNING);
                 case FAIL ->
-                    executor.withException(new IllegalStateException("Flow is FAILED due to concurrency limit exceeded"), "checkConcurrencyLimit");
+                    executionRunning
+                        .withExecution(executionRunning.getExecution().failedExecutionFromExecutor(new IllegalStateException("Execution is FAILED due to concurrency limit exceeded")).getExecution())
+                        .withConcurrencyState(ExecutionRunning.ConcurrencyState.RUNNING);
+
             };
         }
 
-        // if under the limit, update the executor with a RUNNING ExecutionRunning to track them
-        var executionRunning = new ExecutionRunning(
-            flow.getTenantId(),
-            flow.getNamespace(),
-            flow.getId(),
-            executor.getExecution(),
-            ExecutionRunning.ConcurrencyState.RUNNING
-        );
-        return executor.withExecutionRunning(executionRunning);
+        // if under the limit, run it!
+        return executionRunning
+            .withExecution(executionRunning.getExecution().withState(State.Type.RUNNING))
+            .withConcurrencyState(ExecutionRunning.ConcurrencyState.RUNNING);
     }
 
     public Executor process(Executor executor) {
@@ -260,8 +252,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 +729,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 +813,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 +882,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 +927,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 +936,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 +1163,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 +1333,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/FlowableUtils.java

@@ -286,18 +286,10 @@ public class FlowableUtils {
 
         // start as many tasks as we have concurrency slots
         return collect.values().stream()
-            .map(resolvedTasks -> filterCreated(resolvedTasks, taskRuns, parentTaskRun))
+            .map(resolvedTasks -> resolveSequentialNexts(execution, resolvedTasks, null, null, parentTaskRun))
             .filter(resolvedTasks -> !resolvedTasks.isEmpty())
             .limit(concurrencySlots)
-            .map(resolvedTasks -> resolvedTasks.getFirst().toNextTaskRun(execution))
-            .toList();
-    }
-
-    private static List<ResolvedTask> filterCreated(List<ResolvedTask> tasks, List<TaskRun> taskRuns, TaskRun parentTaskRun) {
-        return tasks.stream()
-            .filter(resolvedTask -> taskRuns.stream()
-                .noneMatch(taskRun -> FlowableUtils.isTaskRunFor(resolvedTask, taskRun, parentTaskRun))
-            )
+            .map(resolvedTasks -> resolvedTasks.getFirst())
             .toList();
     }
 

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;
+}