fix(executions): make state_duration generated on queries

- fixes https://github.com/kestra-io/kestra/issues/11593

.github/actions/plugins-list/action.yml

@@ -1,29 +0,0 @@
-name: 'Load Kestra Plugin List'
-description: 'Composite action to load list of plugins'
-inputs:
-  plugin-version:
-    description: "Kestra version"
-    default: 'LATEST'
-    required: true
-  plugin-file:
-    description: "File of the plugins"
-    default: './.plugins'
-    required: true
-outputs:
-  plugins:
-    description: "List of all Kestra plugins"
-    value: ${{ steps.plugins.outputs.plugins }}
-  repositories:
-    description: "List of all Kestra repositories of plugins"
-    value: ${{ steps.plugins.outputs.repositories }}
-runs:
-  using: composite
-  steps:
-    - name: Get Plugins List
-      id: plugins
-      shell: bash
-      run: |
-        PLUGINS=$([ -f ${{ inputs.plugin-file }} ] && cat ${{ inputs.plugin-file }} | grep "io\\.kestra\\." | sed -e '/#/s/^.//' | sed -e "s/LATEST/${{ inputs.plugin-version }}/g" | cut -d':' -f2- | xargs || echo '');
-        REPOSITORIES=$([ -f ${{ inputs.plugin-file }} ] && cat ${{ inputs.plugin-file }} | grep "io\\.kestra\\." | sed -e '/#/s/^.//' | cut -d':' -f1 | uniq | sort | xargs || echo '')
-        echo "plugins=$PLUGINS" >> $GITHUB_OUTPUT
-        echo "repositories=$REPOSITORIES" >> $GITHUB_OUTPUT

.github/actions/setup-vars/action.yml

@@ -1,20 +0,0 @@
-name: 'Setup vars'
-description: 'Composite action to setup common vars'
-outputs:
-  tag:
-    description: "Git tag"
-    value: ${{ steps.vars.outputs.tag }}
-  commit:
-    description: "Git commit"
-    value: ${{ steps.vars.outputs.commit }}
-runs:
-  using: composite
-  steps:
-    # Setup vars
-    - name: Set variables
-      id: vars
-      shell: bash
-      run: |
-        TAG=${GITHUB_REF#refs/*/}
-        echo "tag=${TAG}" >> $GITHUB_OUTPUT
-        echo "commit=$(git rev-parse --short "$GITHUB_SHA")" >> $GITHUB_OUTPUT

.github/workflows/e2e-scheduling.yml

@@ -0,0 +1,15 @@
+name: 'E2E tests scheduling'
+# 'New E2E tests implementation started by Roman. Based on playwright in npm UI project, tests Kestra OSS develop docker image. These tests are written from zero, lets make them unflaky from the start!.'
+on:
+  schedule:
+    - cron: "0 * * * *"  # Every hour
+  workflow_dispatch:
+    inputs:
+      noInputYet:
+        description: 'not input yet.'
+        required: false
+        type: string
+        default: "no input"
+jobs:
+  e2e:
+    uses: kestra-io/actions/.github/workflows/kestra-oss-e2e-tests.yml@main

.github/workflows/e2e.yml

@@ -1,86 +0,0 @@
-name: 'E2E tests revival'
-description: 'New E2E tests implementation started by Roman. Based on playwright in npm UI project, tests Kestra OSS develop docker image. These tests are written from zero, lets make them unflaky from the start!.'
-on:
-  schedule:
-    - cron: "0 * * * *"  # Every hour
-  workflow_call:
-    inputs:
-      noInputYet:
-        description: 'not input yet.'
-        required: false
-        type: string
-        default: "no input"
-  workflow_dispatch:
-    inputs:
-      noInputYet:
-        description: 'not input yet.'
-        required: false
-        type: string
-        default: "no input"
-jobs:
-  check:
-    timeout-minutes: 15
-    runs-on: ubuntu-latest
-    env:
-      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-    steps:
-      - name: Login to DockerHub
-        uses: docker/login-action@v3
-        with:
-          registry: ghcr.io
-          username: ${{ github.actor }}
-          password: ${{ github.token }}
-
-      - name: Checkout kestra
-        uses: actions/checkout@v5
-        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
-          npm i
-          npx playwright install --with-deps chromium
-
-      - name: Run E2E Tests
-        run: |
-          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'
-        uses: actions/upload-artifact@v4
-        if: ${{ !cancelled() }}
-        with:
-          name: playwright-report
-          path: kestra/ui/playwright-report/
-          retention-days: 7
-        # Allure check
-        # TODO I don't know what it should do
-#      - uses: rlespinasse/github-slug-action@v5
-#        name: Allure - Generate slug variables
-#
-#      - name: Allure - Publish report
-#        uses: andrcuns/allure-publish-action@v2.9.0
-#        if: always() && env.GOOGLE_SERVICE_ACCOUNT != ''
-#        continue-on-error: true
-#        env:
-#          GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
-#          JAVA_HOME: /usr/lib/jvm/default-jvm/
-#        with:
-#          storageType: gcs
-#          resultsGlob: "**/build/allure-results"
-#          bucket: internal-kestra-host
-#          baseUrl: "https://internal.dev.kestra.io"
-#          prefix: ${{ format('{0}/{1}', github.repository, 'allure/java') }}
-#          copyLatest: true
-#          ignoreMissingResults: true

.github/workflows/global-create-new-release-branch.yml

@@ -1,5 +1,5 @@
-name: Run Gradle Release
-run-name: "Releasing Kestra ${{ github.event.inputs.releaseVersion }} 🚀"
+name: Create new release branch
+run-name: "Create new release branch Kestra ${{ github.event.inputs.releaseVersion }} 🚀"
 on:
   workflow_dispatch:
     inputs:
@@ -38,15 +38,8 @@ jobs:
           fetch-depth: 0
           path: kestra
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true

.github/workflows/global-gradle-release-plugins.yml

@@ -25,25 +25,17 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true
           node-enabled: true
           python-enabled: true
-          caches-enabled: true
 
       # Get Plugins List
       - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
+        uses: kestra-io/actions/composite/kestra-oss/kestra-oss-plugins-list@main
         id: plugins-list
         with:
           plugin-version: 'LATEST'
@@ -60,7 +52,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/release-plugins.sh;
-          
+
           ./dev-tools/release-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --next-version=${{github.event.inputs.nextVersion}} \
@@ -73,7 +65,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/release-plugins.sh;
-          
+
           ./dev-tools/release-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --next-version=${{github.event.inputs.nextVersion}} \

.github/workflows/global-setversion-tag-plugins.yml

@@ -23,7 +23,7 @@ jobs:
 
       # Get Plugins List
       - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
+        uses: kestra-io/actions/composite/kestra-oss/kestra-oss-plugins-list@main
         id: plugins-list
         with:
           plugin-version: 'LATEST'
@@ -40,7 +40,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/setversion-tag-plugins.sh;
-          
+
           ./dev-tools/setversion-tag-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --yes \
@@ -52,7 +52,7 @@ jobs:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |
           chmod +x ./dev-tools/setversion-tag-plugins.sh;
-          
+
           ./dev-tools/setversion-tag-plugins.sh \
           --release-version=${{github.event.inputs.releaseVersion}} \
           --dry-run \

.github/workflows/global-start-release.yml

@@ -1,5 +1,5 @@
-name: Set Version and Tag
-run-name: "Set version and Tag Kestra to ${{ github.event.inputs.releaseVersion }} 🚀"
+name: Start release
+run-name: "Start release of Kestra ${{ github.event.inputs.releaseVersion }} 🚀"
 on:
   workflow_dispatch:
     inputs:
@@ -7,17 +7,26 @@ on:
         description: 'The release version (e.g., 0.21.1)'
         required: true
         type: string
+
+permissions:
+  contents: write
+
 env:
   RELEASE_VERSION: "${{ github.event.inputs.releaseVersion }}"
 jobs:
   release:
     name: Release Kestra
     runs-on: ubuntu-latest
-    if: startsWith(github.ref, 'refs/heads/releases/v')
     steps:
-      # Checks
-      - name: Check Inputs
+      - name: Parse and Check Inputs
+        id: parse-and-check-inputs
         run: |
+          CURRENT_BRANCH="${{ github.ref_name }}"
+          if ! [[ "$CURRENT_BRANCH" == "develop" ]]; then
+            echo "You can only run this workflow on develop, but you ran it on $CURRENT_BRANCH"
+            exit 1
+          fi
+
           if ! [[ "$RELEASE_VERSION" =~ ^[0-9]+(\.[0-9]+)(\.[0-9]+)(-rc[0-9])?(-SNAPSHOT)?$ ]]; then
             echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)(\.[0-9]+)-(rc[0-9])?(-SNAPSHOT)?$"
             exit 1
@@ -25,13 +34,8 @@ jobs:
 
           # Extract the major and minor versions
           BASE_VERSION=$(echo "$RELEASE_VERSION" | sed -E 's/^([0-9]+\.[0-9]+)\..*/\1/')
-          RELEASE_BRANCH="refs/heads/releases/v${BASE_VERSION}.x"
-
-          CURRENT_BRANCH="$GITHUB_REF"
-          if ! [[ "$CURRENT_BRANCH" == "$RELEASE_BRANCH" ]]; then
-            echo "Invalid release branch. Expected $RELEASE_BRANCH, was $CURRENT_BRANCH"
-            exit 1
-          fi
+          RELEASE_BRANCH="releases/v${BASE_VERSION}.x"
+          echo "release_branch=${RELEASE_BRANCH}" >> $GITHUB_OUTPUT
 
       # Checkout
       - name: Checkout
@@ -39,6 +43,7 @@ jobs:
         with:
           fetch-depth: 0
           token: ${{ secrets.GH_PERSONAL_TOKEN }}
+          ref: ${{ steps.parse-and-check-inputs.outputs.release_branch }}
 
       # Configure
       - name: Git - Configure
@@ -47,7 +52,7 @@ jobs:
           git config --global user.name "github-actions[bot]"
 
       # Execute
-      - name: Run Gradle Release
+      - name: Start release by updating version and pushing a new tag
         env:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
         run: |

.github/workflows/main-build.yml

@@ -0,0 +1,86 @@
+name: Main Workflow
+
+on:
+  push:
+    branches:
+      - releases/*
+      - develop
+
+  workflow_dispatch:
+    inputs:
+      skip-test:
+        description: 'Skip test'
+        type: choice
+        required: true
+        default: 'false'
+        options:
+          - "true"
+          - "false"
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}-main
+  cancel-in-progress: true
+
+jobs:
+  backend-tests:
+    name: Backend tests
+    if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
+    uses: kestra-io/actions/.github/workflows/kestra-oss-backend-tests.yml@main
+    secrets:
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
+
+  frontend-tests:
+    name: Frontend tests
+    if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
+    uses: kestra-io/actions/.github/workflows/kestra-oss-frontend-tests.yml@main
+    secrets:
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+  publish-develop-docker:
+    name: Publish Docker
+    needs: [backend-tests, frontend-tests]
+    if: "!failure() && !cancelled() && github.ref == 'refs/heads/develop'"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-docker.yml@main
+    with:
+      plugin-version: 'LATEST-SNAPSHOT'
+    secrets:
+      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
+      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
+      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+
+
+  publish-develop-maven:
+    name: Publish develop Maven
+    needs: [ backend-tests, frontend-tests ]
+    if: "!failure() && !cancelled() && github.ref == 'refs/heads/develop'"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-maven.yml@main
+    secrets:
+      SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
+      SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
+      SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
+      SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
+      SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
+
+  end:
+    runs-on: ubuntu-latest
+    needs: [publish-develop-docker, publish-develop-maven]
+    if: always()
+    steps:
+      - name: Trigger EE Workflow
+        uses: peter-evans/repository-dispatch@v3
+        if: github.ref == 'refs/heads/develop' && needs.release.result == 'success'
+        with:
+          token: ${{ secrets.GH_PERSONAL_TOKEN }}
+          repository: kestra-io/kestra-ee
+          event-type: "oss-updated"
+
+      # Slack
+      - name: Slack - Notification
+        if: ${{ failure() && env.SLACK_WEBHOOK_URL != 0 && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/main' || github.ref == 'refs/heads/develop') }}
+        uses: kestra-io/actions/composite/slack-status@main
+        with:
+          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}

.github/workflows/main.yml

@@ -1,83 +0,0 @@
-name: Main Workflow
-
-on:
-  workflow_dispatch:
-    inputs:
-      skip-test:
-        description: 'Skip test'
-        type: choice
-        required: true
-        default: 'false'
-        options:
-          - "true"
-          - "false"
-      plugin-version:
-          description: "plugins version"
-          required: false
-          type: string
-  push:
-    branches:
-      - master
-      - main
-      - releases/*
-      - develop
-    tags:
-      - v*
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}-main
-  cancel-in-progress: true
-
-jobs:
-  tests:
-    name: Execute tests
-    uses: ./.github/workflows/workflow-test.yml
-    if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
-    with:
-      report-status: false
-
-  release:
-    name: Release
-    needs: [tests]
-    if: "!failure() && !cancelled() && !startsWith(github.ref, 'refs/heads/releases')"
-    uses: ./.github/workflows/workflow-release.yml
-    with:
-        plugin-version: ${{ inputs.plugin-version != '' && inputs.plugin-version || (github.ref == 'refs/heads/develop' && 'LATEST-SNAPSHOT' || 'LATEST') }}
-    secrets:
-      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
-      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
-      SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
-      SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
-      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:
-      - release
-    if: always()
-    env:
-      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
-    steps:
-      - name: Trigger EE Workflow
-        uses: peter-evans/repository-dispatch@v3
-        if: github.ref == 'refs/heads/develop' && needs.release.result == 'success'
-        with:
-          token: ${{ secrets.GH_PERSONAL_TOKEN }}
-          repository: kestra-io/kestra-ee
-          event-type: "oss-updated"
-
-
-      # Slack
-      - name: Slack - Notification
-        uses: Gamesight/slack-workflow-status@master
-        if: ${{ always() && env.SLACK_WEBHOOK_URL != 0 }}
-        with:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ":github-actions:"
-          channel: "C02DQ1A7JLR" # _int_git channel

.github/workflows/pre-release.yml

@@ -0,0 +1,49 @@
+name: Pre Release
+
+on:
+  push:
+    tags:
+      - 'v*'
+  workflow_dispatch:
+
+jobs:
+  build-artifacts:
+    name: Build Artifacts
+    uses: kestra-io/actions/.github/workflows/kestra-oss-build-artifacts.yml@main
+
+  backend-tests:
+    name: Backend tests
+    uses: kestra-io/actions/.github/workflows/kestra-oss-backend-tests.yml@main
+    secrets:
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
+      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
+
+  frontend-tests:
+    name: Frontend tests
+    uses: kestra-io/actions/.github/workflows/kestra-oss-frontend-tests.yml@main
+    secrets:
+      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+
+  publish-maven:
+    name: Publish Maven
+    needs: [ backend-tests, frontend-tests ]
+    if: "!failure() && !cancelled()"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-maven.yml@main
+    secrets:
+      SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
+      SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
+      SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
+      SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
+      SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
+
+  publish-github:
+    name: Github Release
+    needs: [build-artifacts, backend-tests, frontend-tests]
+    if: "!failure() && !cancelled()"
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-github.yml@main
+    secrets:
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}

.github/workflows/pull-request-cleanup.yml

@@ -3,7 +3,7 @@ name: Pull Request - Delete Docker
 on:
   pull_request:
     types: [closed]
-
+# TODO import a reusable one
 jobs:
   publish:
     name: Pull Request - Delete Docker

.github/workflows/pull-request.yml

@@ -2,17 +2,12 @@ name: Pull Request Workflow
 
 on:
   pull_request:
-    branches:
-      - develop
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref_name }}-pr
   cancel-in-progress: true
 
 jobs:
-  # ********************************************************************************************************************
-  # File changes detection
-  # ********************************************************************************************************************
   file-changes:
     if: ${{ github.event.pull_request.draft == false }}
     name: File changes detection
@@ -33,14 +28,11 @@ jobs:
               - '!{ui,.github}/**'
           token: ${{ secrets.GITHUB_TOKEN }}
 
-  # ********************************************************************************************************************
-  # Tests
-  # ********************************************************************************************************************
   frontend:
     name: Frontend - Tests
     needs: [file-changes]
     if: "needs.file-changes.outputs.ui == 'true'"
-    uses: ./.github/workflows/workflow-frontend-test.yml
+    uses: kestra-io/actions/.github/workflows/kestra-oss-frontend-tests.yml@main
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
@@ -49,7 +41,7 @@ jobs:
     name: Backend - Tests
     needs: file-changes
     if: "needs.file-changes.outputs.backend == 'true'"
-    uses: ./.github/workflows/workflow-backend-test.yml
+    uses: kestra-io/actions/.github/workflows/kestra-oss-backend-tests.yml@main
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
@@ -58,21 +50,8 @@ jobs:
 
   e2e-tests:
     name: E2E - Tests
-    uses: ./.github/workflows/e2e.yml
+    uses: kestra-io/actions/.github/workflows/kestra-oss-e2e-tests.yml@main
 
-  end:
-    name: End
-    runs-on: ubuntu-latest
-    if: always()
-    needs: [frontend, backend]
-    steps:
-      # Slack
-      - name: Slack notification
-        uses: Gamesight/slack-workflow-status@master
-        if: ${{ always() && env.SLACK_WEBHOOK_URL != 0 }}
-        with:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ":github-actions:"
-          channel: "C02DQ1A7JLR"
+  generate-pull-request-docker-image:
+    name: Generate PR docker image
+    uses: kestra-io/actions/.github/workflows/kestra-oss-pullrequest-publish-docker.yml@main

.github/workflows/release-docker.yml

@@ -0,0 +1,34 @@
+name: Publish docker
+
+on:
+  workflow_dispatch:
+    inputs:
+      retag-latest:
+        description: 'Retag latest Docker images'
+        required: true
+        type: boolean
+        default: false
+      retag-lts:
+        description: 'Retag LTS Docker images'
+        required: true
+        type: boolean
+        default: false
+      plugin-version:
+        description: 'Plugin version'
+        required: false
+        type: string
+        default: "LATEST"
+
+jobs:
+  publish-docker:
+    name: Publish Docker
+    if: startsWith(github.ref, 'refs/tags/v')
+    uses: kestra-io/actions/.github/workflows/kestra-oss-publish-docker.yml@main
+    with:
+      plugin-version: ${{ inputs.plugin-version }}
+      retag-latest: ${{ inputs.retag-latest }}
+      retag-lts: ${{ inputs.retag-lts }}
+    secrets:
+      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
+      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
+      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

.github/workflows/vulnerabilities-check.yml

@@ -21,13 +21,6 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
       - uses: ./actions/.github/actions/setup-build
         id: build
@@ -70,15 +63,8 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: false
@@ -115,20 +101,12 @@ jobs:
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
-      - uses: ./actions/.github/actions/setup-build
+      - uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: false
           node-enabled: false
-          caches-enabled: true
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check

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

@@ -1,142 +0,0 @@
-name: Backend - Tests
-
-on:
-  workflow_call:
-    secrets:
-      GITHUB_AUTH_TOKEN:
-        description: "The GitHub Token."
-        required: true
-      CODECOV_TOKEN:
-        description: 'Codecov Token'
-        required: true
-      SONAR_TOKEN:
-        description: 'Sonar Token'
-        required: true
-      GOOGLE_SERVICE_ACCOUNT:
-        description: 'Google Service Account'
-        required: true
-
-permissions:
-  contents: write
-  checks: write
-  actions: read
-
-jobs:
-  test:
-    name: Backend - Tests
-    runs-on: ubuntu-latest
-    env:
-      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-    steps:
-      - uses: actions/checkout@v5
-        name: Checkout - Current ref
-        with:
-          fetch-depth: 0
-
-      # 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
-
-      # Services
-      - name: Setup - Start docker compose
-        shell: bash
-        run: docker compose -f docker-compose-ci.yml up -d
-
-      # Gradle check
-      - name: Gradle - Build
-        if: ${{ github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '' }}
-        env:
-          GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-        shell: bash
-        run: |
-          echo $GOOGLE_SERVICE_ACCOUNT | base64 -d > ~/.gcp-service-account.json
-          export GOOGLE_APPLICATION_CREDENTIALS=$HOME/.gcp-service-account.json
-          ./gradlew check javadoc --parallel
-
-      # report test
-      - name: Test - Publish Test Results
-        uses: dorny/test-reporter@v2
-        if: always()
-        with:
-          name: Java Tests Report
-          reporter: java-junit
-          path: '**/build/test-results/test/TEST-*.xml'
-          list-suites: 'failed'
-          list-tests: 'failed'
-          fail-on-error: 'false'
-          token: ${{ secrets.GITHUB_AUTH_TOKEN }}
-
-      # Sonar
-      - name: Test - Analyze with Sonar
-        if: env.SONAR_TOKEN != ''
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
-          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-        shell: bash
-        run: ./gradlew sonar --info
-
-      # GCP
-      - name: GCP - Auth with unit test account
-        id: auth
-        if: always() && env.GOOGLE_SERVICE_ACCOUNT != ''
-        continue-on-error: true
-        uses: "google-github-actions/auth@v3"
-        with:
-          credentials_json: "${{ secrets.GOOGLE_SERVICE_ACCOUNT }}"
-
-      - name: GCP - Setup Cloud SDK
-        if: env.GOOGLE_SERVICE_ACCOUNT != ''
-        uses: "google-github-actions/setup-gcloud@v3"
-
-        # Allure check
-      - uses: rlespinasse/github-slug-action@v5
-        name: Allure - Generate slug variables
-
-      - name: Allure - Publish report
-        uses: andrcuns/allure-publish-action@v2.9.0
-        if: always() && env.GOOGLE_SERVICE_ACCOUNT != ''
-        continue-on-error: true
-        env:
-          GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
-          JAVA_HOME: /usr/lib/jvm/default-jvm/
-        with:
-          storageType: gcs
-          resultsGlob: "**/build/allure-results"
-          bucket: internal-kestra-host
-          baseUrl: "https://internal.dev.kestra.io"
-          prefix: ${{ format('{0}/{1}', github.repository, 'allure/java') }}
-          copyLatest: true
-          ignoreMissingResults: true
-
-      # Jacoco
-      - name: Jacoco - Copy reports
-        if: env.GOOGLE_SERVICE_ACCOUNT != ''
-        continue-on-error: true
-        shell: bash
-        run: |
-          mv build/reports/jacoco/testCodeCoverageReport build/reports/jacoco/test/
-          mv build/reports/jacoco/test/testCodeCoverageReport.xml build/reports/jacoco/test/jacocoTestReport.xml
-          gsutil -m rsync -d -r  build/reports/jacoco/test/ gs://internal-kestra-host/${{ format('{0}/{1}', github.repository, 'jacoco') }}
-
-      # Codecov
-      - name: Codecov - Upload coverage reports
-        uses: codecov/codecov-action@v5
-        if: ${{ !cancelled() }}
-        continue-on-error: true
-        with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          flags: backend
-
-      - name: Codecov - Upload test results
-        uses: codecov/test-results-action@v1
-        if: ${{ !cancelled() }}
-        continue-on-error: true
-        with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          flags: backend

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

@@ -1,80 +0,0 @@
-name: Build Artifacts
-
-on:
-  workflow_call: {}
-
-jobs:
-  build:
-    name: Build - Artifacts
-    runs-on: ubuntu-latest
-    outputs:
-        docker-tag: ${{ steps.vars.outputs.tag }}
-        docker-artifact-name: ${{ steps.vars.outputs.artifact }}
-        plugins: ${{ steps.plugins.outputs.plugins }}
-    env:
-      PLUGIN_VERSION: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
-    steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v5
-        with:
-          fetch-depth: 0
-
-      # Npm
-      - name: Setup - Npm install
-        shell: bash
-        working-directory: ui
-        run: npm ci
-
-      # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
-        name: Setup - Build
-        id: build
-        with:
-          java-enabled: true
-          node-enabled: true
-
-      # Get Plugins List
-      - name: Plugins - Get List
-        uses: ./.github/actions/plugins-list
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        id: plugins-list
-        with:
-          plugin-version: ${{ env.PLUGIN_VERSION }}
-
-      # Set Plugins List
-      - name: Plugins - Set List
-        id: plugins
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        shell: bash
-        run: |
-          PLUGINS="${{ steps.plugins-list.outputs.plugins }}"
-          TAG=${GITHUB_REF#refs/*/}
-          if [[ $TAG = "master" || $TAG == v* ]]; then
-            echo "plugins=$PLUGINS" >> $GITHUB_OUTPUT
-          else
-            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
-          fi
-
-      # Build
-      - name: Gradle - Build
-        shell: bash
-        run: |
-          ./gradlew executableJar
-
-      - name: Artifacts - Copy exe to image
-        shell: bash
-        run: |
-          cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
-
-      # Upload artifacts
-      - name: Artifacts - Upload JAR
-        uses: actions/upload-artifact@v4
-        with:
-          name: jar
-          path: build/libs/
-
-      - name: Artifacts - Upload Executable
-        uses: actions/upload-artifact@v4
-        with:
-          name: exe
-          path: build/executable/

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

@@ -1,70 +0,0 @@
-name: Frontend - Tests
-
-on:
-  workflow_call:
-    secrets:
-      GITHUB_AUTH_TOKEN:
-        description: "The GitHub Token."
-        required: true
-      CODECOV_TOKEN:
-        description: 'Codecov Token'
-        required: true
-
-env:
-  # to save corepack from itself
-  COREPACK_INTEGRITY_KEYS: 0
-
-jobs:
-  test:
-    name: Frontend - Tests
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v5
-
-      - name: Cache Node Modules
-        id: cache-node-modules
-        uses: actions/cache@v4
-        with:
-          path: |
-            ui/node_modules
-          key: modules-${{ hashFiles('ui/package-lock.json') }}
-
-      - name: Cache Playwright Binaries
-        id: cache-playwright
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/ms-playwright
-          key: playwright-${{ hashFiles('ui/package-lock.json') }}
-
-      - name: Npm - install
-        if: steps.cache-node-modules.outputs.cache-hit != 'true'
-        working-directory: ui
-        run: npm ci
-
-      - name: Npm - lint
-        uses: reviewdog/action-eslint@v1
-        with:
-          github_token: ${{ secrets.GITHUB_AUTH_TOKEN }}
-          reporter: github-pr-review
-          workdir: ui
-
-      - name: Npm - Run build
-        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
-        working-directory: ui
-        if: steps.cache-playwright.outputs.cache-hit != 'true'
-        run: npx playwright install --with-deps
-
-      - name: Run storybook component tests
-        working-directory: ui
-        run: npm run test:storybook -- --coverage

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

@@ -1,88 +0,0 @@
-name: Github - Release
-
-on:
-  workflow_dispatch:
-  workflow_call:
-    secrets:
-      GH_PERSONAL_TOKEN:
-        description: "The Github personal token."
-        required: true
-      SLACK_RELEASES_WEBHOOK_URL:
-        description: "The Slack webhook URL."
-        required: true
-
-
-
-jobs:
-  publish:
-    name: Github - Release
-    runs-on: ubuntu-latest
-    steps:
-      # Check out
-      - name: Checkout - Repository
-        uses: actions/checkout@v5
-        with:
-          fetch-depth: 0
-          submodules: true
-
-      # Checkout GitHub Actions
-      - name: Checkout - Actions
-        uses: actions/checkout@v5
-        with:
-          repository: kestra-io/actions
-          sparse-checkout-cone-mode: true
-          path: actions
-          sparse-checkout: |
-            .github/actions
-
-      # Download Exec
-      # Must be done after checkout actions
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v5
-        if: startsWith(github.ref, 'refs/tags/v')
-        with:
-          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 }}
-
-      # Trigger gha workflow to bump helm chart version
-      - name: GitHub - Trigger the Helm chart version bump
-        uses: peter-evans/repository-dispatch@v3
-        with:
-          token: ${{ secrets.GH_PERSONAL_TOKEN }}
-          repository: kestra-io/helm-charts
-          event-type: update-helm-chart-version
-          client-payload: |-
-            {
-              "new_version": "${{ github.ref_name }}",
-              "github_repository": "${{ github.repository }}",
-              "github_actor": "${{ github.actor }}"
-            }
-
-      - name: Merge Release Notes
-        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
-        uses: ./actions/.github/actions/github-release-note-merge
-        env:
-            GITHUB_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
-            RELEASE_TAG: ${{ github.ref_name }}

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

@@ -1,213 +0,0 @@
-name: Create Docker images on Release
-
-on:
-  workflow_dispatch:
-    inputs:
-      retag-latest:
-        description: 'Retag latest Docker images'
-        required: true
-        type: choice
-        default: "false"
-        options:
-          - "true"
-          - "false"
-      retag-lts:
-        description: 'Retag LTS Docker images'
-        required: true
-        type: choice
-        default: "false"
-        options:
-          - "true"
-          - "false"
-      release-tag:
-        description: 'Kestra Release Tag (by default, deduced with the ref)'
-        required: false
-        type: string
-      plugin-version:
-        description: 'Plugin version'
-        required: false
-        type: string
-        default: "LATEST"
-      force-download-artifact:
-        description: 'Force download artifact'
-        required: false
-        type: choice
-        default: "true"
-        options:
-          - "true"
-          - "false"
-  workflow_call:
-    inputs:
-      plugin-version:
-        description: "Plugin version"
-        default: 'LATEST'
-        required: false
-        type: string
-      force-download-artifact:
-        description: 'Force download artifact'
-        required: false
-        type: string
-        default: "true"
-    secrets:
-      DOCKERHUB_USERNAME:
-        description: "The Dockerhub username."
-        required: true
-      DOCKERHUB_PASSWORD:
-        description: "The Dockerhub password."
-        required: true
-
-env:
-  PLUGIN_VERSION: ${{ inputs.plugin-version != null && inputs.plugin-version || 'LATEST' }}
-jobs:
-  plugins:
-    name: List Plugins
-    runs-on: ubuntu-latest
-    outputs:
-      plugins: ${{ steps.plugins.outputs.plugins }}
-    steps:
-      # Checkout
-      - uses: actions/checkout@v5
-
-      # Get Plugins List
-      - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
-        id: plugins
-        with: # remap LATEST-SNAPSHOT to LATEST
-          plugin-version: ${{ env.PLUGIN_VERSION == 'LATEST-SNAPSHOT' && 'LATEST' || env.PLUGIN_VERSION }}
-
-  # ********************************************************************************************************************
-  # Build
-  # ********************************************************************************************************************
-  build-artifacts:
-    name: Build Artifacts
-    if: ${{ inputs.force-download-artifact == 'true' }}
-    uses: ./.github/workflows/workflow-build-artifacts.yml
-
-  docker:
-    name: Publish Docker
-    needs: [ plugins, build-artifacts ]
-    if: always()
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        image:
-          - name: "-no-plugins"
-            plugins: ""
-            packages: jattach
-            python-libs: ""
-          - name: ""
-            plugins: ${{needs.plugins.outputs.plugins}}
-            packages: python3 python-is-python3 python3-pip curl jattach
-            python-libs: kestra
-    steps:
-      - uses: actions/checkout@v5
-
-      # Vars
-      - name: Set image name
-        id: vars
-        run: |
-          if [[ "${{ inputs.release-tag }}" == "" ]]; then
-            TAG=${GITHUB_REF#refs/*/}
-            echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          else
-            TAG="${{ inputs.release-tag }}"
-            echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          fi
-
-          if [[ $GITHUB_REF == refs/tags/* ]]; then
-            if [[ $TAG =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
-              # this will remove the patch version number
-              MINOR_SEMVER=${TAG%.*}
-              echo "minor_semver=${MINOR_SEMVER}" >> $GITHUB_OUTPUT
-            else
-              echo "Tag '$TAG' is not a valid semver (vMAJOR.MINOR.PATCH), skipping minor_semver"
-            fi
-          fi
-
-          if [[ "${{ env.PLUGIN_VERSION }}" == *"-SNAPSHOT" ]]; then
-            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 executable from artifact
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v5
-        with:
-          name: exe
-          path: build/executable
-
-      - name: Copy exe to image
-        run: |
-          cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
-
-      # Docker setup
-      - name: Set up 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: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
-
-      # Docker Login
-      - name: Login to DockerHub
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_PASSWORD }}
-
-      # Docker Build and push
-      - name: Push to Docker Hub
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          push: true
-          tags: ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }}
-          platforms: linux/amd64,linux/arm64
-          build-args: |
-            KESTRA_PLUGINS=${{ steps.vars.outputs.plugins }}
-            APT_PACKAGES=${{ matrix.image.packages }}
-            PYTHON_LIBRARIES=${{ matrix.image.python-libs }}
-
-      - name: Install regctl
-        if: startsWith(github.ref, 'refs/tags/v')
-        uses: regclient/actions/regctl-installer@main
-
-      - name: Retag to minor semver version
-        if: startsWith(github.ref, 'refs/tags/v') && steps.vars.outputs.minor_semver != ''
-        run: |
-          regctl image copy ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }} ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.minor_semver, matrix.image.name) }}
-
-      - name: Retag to latest
-        if: startsWith(github.ref, 'refs/tags/v') && inputs.retag-latest == 'true'
-        run: |
-          regctl image copy ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }} ${{ format('kestra/kestra:latest{0}', matrix.image.name) }}
-
-      - name: Retag to LTS
-        if: startsWith(github.ref, 'refs/tags/v') && inputs.retag-lts == 'true'
-        run: |
-          regctl image copy ${{ format('kestra/kestra:{0}{1}', steps.vars.outputs.tag, matrix.image.name) }} ${{ format('kestra/kestra:latest-lts{0}', matrix.image.name) }}
-
-  end:
-    runs-on: ubuntu-latest
-    needs:
-      - docker
-    if: always()
-    env:
-      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
-    steps:
-
-      # Slack
-      - name: Slack notification
-        uses: Gamesight/slack-workflow-status@master
-        if: ${{ always() && env.SLACK_WEBHOOK_URL != 0 }}
-        with:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ':github-actions:'
-          channel: 'C02DQ1A7JLR' # _int_git channel

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

@@ -1,57 +0,0 @@
-name: Publish - Maven
-
-on:
-  workflow_call:
-    secrets:
-      SONATYPE_USER:
-        description: "The Sonatype username."
-        required: true
-      SONATYPE_PASSWORD:
-        description: "The Sonatype password."
-        required: true
-      SONATYPE_GPG_KEYID:
-        description: "The Sonatype GPG key id."
-        required: true
-      SONATYPE_GPG_PASSWORD:
-        description: "The Sonatype GPG password."
-        required: true
-      SONATYPE_GPG_FILE:
-        description: "The Sonatype GPG file."
-        required: true
-
-jobs:
-  publish:
-    name: Publish - Maven
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v5
-
-      # Setup build
-      - name: Setup - Build
-        uses: kestra-io/actions/.github/actions/setup-build@main
-        id: build
-        with:
-          java-enabled: true
-          node-enabled: true
-
-      # Publish
-      - name: Publish - Release package to Maven Central
-        shell: bash
-        env:
-          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}}
-        run: |
-          mkdir -p ~/.gradle/
-          echo "signing.keyId=${SONATYPE_GPG_KEYID}" > ~/.gradle/gradle.properties
-          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 publishToMavenCentral
-
-      # Gradle dependency
-      - name: Java - Gradle dependency graph
-        uses: gradle/actions/dependency-submission@v4

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

@@ -1,78 +0,0 @@
-name: Pull Request - Publish Docker
-
-on:
-  pull_request:
-    branches:
-      - develop
-
-jobs:
-  build-artifacts:
-    name: Build Artifacts
-    if: github.repository == 'kestra-io/kestra' # prevent running on forks
-    uses: ./.github/workflows/workflow-build-artifacts.yml
-
-  publish:
-    name: Publish Docker
-    if: github.repository == 'kestra-io/kestra' # prevent running on forks
-    runs-on: ubuntu-latest
-    needs: build-artifacts
-    env:
-      GITHUB_IMAGE_PATH: "ghcr.io/kestra-io/kestra-pr"
-    steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v5
-        with:
-          fetch-depth: 0
-
-      # Docker setup
-      - name: Docker - Setup QEMU
-        uses: docker/setup-qemu-action@v3
-
-      - name: Docker - Setup Docker Buildx
-        uses: docker/setup-buildx-action@v3
-
-      # Docker Login
-      - name: Login to GHCR
-        uses: docker/login-action@v3
-        with:
-          registry: ghcr.io
-          username: ${{ github.actor }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      # Build Docker Image
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v5
-        with:
-          name: exe
-          path: build/executable
-
-      - name: Docker - Copy exe to image
-        shell: bash
-        run: |
-          cp build/executable/* docker/app/kestra && chmod +x docker/app/kestra
-
-      - name: Docker - Build image
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          file: ./Dockerfile.pr
-          push: true
-          tags: ${{ env.GITHUB_IMAGE_PATH }}:${{ github.event.pull_request.number }}
-          platforms: linux/amd64,linux/arm64
-
-      # Add comment on pull request
-      - name: Add comment to PR
-        uses: actions/github-script@v8
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          script: |
-            await github.rest.issues.createComment({
-              issue_number: context.issue.number,
-              owner: context.repo.owner,
-              repo: context.repo.repo,
-              body: `**🐋 Docker image**: \`${{ env.GITHUB_IMAGE_PATH }}:${{ github.event.pull_request.number }}\`\n` +
-                `\n` +
-                `\`\`\`bash\n` +
-                `docker run --pull=always --rm -it -p 8080:8080 --user=root -v /var/run/docker.sock:/var/run/docker.sock -v /tmp:/tmp ${{ env.GITHUB_IMAGE_PATH }}:${{ github.event.pull_request.number }} server local\n` +
-                `\`\`\``
-            })

.github/workflows/workflow-release.yml

@@ -1,85 +0,0 @@
-name: Release
-
-on:
-  workflow_dispatch:
-    inputs:
-      plugin-version:
-        description: "plugins version"
-        default: 'LATEST'
-        required: false
-        type: string
-      publish-docker:
-        description: "Publish Docker image"
-        default: 'false'
-        required: false
-        type: string
-  workflow_call:
-    inputs:
-      plugin-version:
-        description: "plugins version"
-        default: 'LATEST'
-        required: false
-        type: string
-    secrets:
-      DOCKERHUB_USERNAME:
-        description: "The Dockerhub username."
-        required: true
-      DOCKERHUB_PASSWORD:
-        description: "The Dockerhub password."
-        required: true
-      SONATYPE_USER:
-        description: "The Sonatype username."
-        required: true
-      SONATYPE_PASSWORD:
-        description: "The Sonatype password."
-        required: true
-      SONATYPE_GPG_KEYID:
-        description: "The Sonatype GPG key id."
-        required: true
-      SONATYPE_GPG_PASSWORD:
-        description: "The Sonatype GPG password."
-        required: true
-      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
-
-  Docker:
-    name: Publish Docker
-    needs: build-artifacts
-    uses: ./.github/workflows/workflow-publish-docker.yml
-    if: github.ref == 'refs/heads/develop' || inputs.publish-docker == 'true'
-    with:
-      force-download-artifact: 'false'
-      plugin-version: ${{ inputs.plugin-version != null && inputs.plugin-version || 'LATEST' }}
-    secrets:
-      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
-      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
-
-  Maven:
-    name: Publish Maven
-    uses: ./.github/workflows/workflow-publish-maven.yml
-    secrets:
-      SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
-      SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
-      SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
-      SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
-      SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
-
-  Github:
-    name: Github Release
-    needs: build-artifacts
-    if: startsWith(github.ref, 'refs/tags/v')
-    uses: ./.github/workflows/workflow-github-release.yml
-    secrets:
-      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
-      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}

.github/workflows/workflow-test.yml

@@ -1,97 +0,0 @@
-name: Tests
-
-on:
-  schedule:
-    - cron: '0 4 * * 1,2,3,4,5'
-  workflow_call:
-    inputs:
-      report-status:
-        description: "Report status of the jobs in outputs"
-        type: string
-        required: false
-        default: false
-    outputs:
-      frontend_status:
-        description: "Status of the frontend job"
-        value: ${{ jobs.set-frontend-status.outputs.frontend_status }}
-      backend_status:
-        description: "Status of the backend job"
-        value: ${{ jobs.set-backend-status.outputs.backend_status }}
-
-jobs:
-  file-changes:
-    name: File changes detection
-    runs-on: ubuntu-latest
-    timeout-minutes: 60
-    outputs:
-      ui: ${{ steps.changes.outputs.ui }}
-      backend: ${{ steps.changes.outputs.backend }}
-    steps:
-      - uses: actions/checkout@v5
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-      - uses: dorny/paths-filter@v3
-        if: "!startsWith(github.ref, 'refs/tags/v')"
-        id: changes
-        with:
-          filters: |
-            ui:
-              - 'ui/**'
-            backend:
-              - '!{ui,.github}/**'
-          token: ${{ secrets.GITHUB_TOKEN }}
-
-  frontend:
-    name: Frontend - Tests
-    needs: file-changes
-    if: "needs.file-changes.outputs.ui == 'true' || startsWith(github.ref, 'refs/tags/v')"
-    uses: ./.github/workflows/workflow-frontend-test.yml
-    secrets:
-      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-
-
-  backend:
-    name: Backend - Tests
-    needs: file-changes
-    if: "needs.file-changes.outputs.backend == 'true' || startsWith(github.ref, 'refs/tags/v')"
-    uses: ./.github/workflows/workflow-backend-test.yml
-    secrets:
-      GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-      SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
-      GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
-
-  # Output every job status
-  # To be used in other workflows
-  report-status:
-    name: Report Status
-    runs-on: ubuntu-latest
-    needs: [ frontend, backend ]
-    if: always() && (inputs.report-status == 'true')
-    outputs:
-      frontend_status: ${{ steps.set-frontend-status.outputs.frontend_status }}
-      backend_status: ${{ steps.set-backend-status.outputs.backend_status }}
-    steps:
-      - id: set-frontend-status
-        name: Set frontend job status
-        run: echo "::set-output name=frontend_status::${{ needs.frontend.result }}"
-
-      - id: set-backend-status
-        name: Set backend job status
-        run: echo "::set-output name=backend_status::${{ needs.backend.result }}"
-
-  notify:
-    name: Notify - Slack
-    runs-on: ubuntu-latest
-    needs: [ frontend, backend ]
-    if: github.event_name == 'schedule'
-    steps:
-      - name: Notify failed CI
-        id: send-ci-failed
-        if: |
-          always() && (needs.frontend.result != 'success' ||
-          needs.backend.result != 'success')
-        uses: kestra-io/actions/.github/actions/send-ci-failed@main
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

README.md

@@ -19,9 +19,12 @@
 <br />
 
 <p align="center">
-    <a href="https://x.com/kestra_io"><img height="25" src="https://kestra.io/twitter.svg" alt="X(formerly Twitter)" /></a> &nbsp;
-    <a href="https://www.linkedin.com/company/kestra/"><img height="25" src="https://kestra.io/linkedin.svg" alt="linkedin" /></a> &nbsp;
-<a href="https://www.youtube.com/@kestra-io"><img height="25" src="https://kestra.io/youtube.svg" alt="youtube" /></a> &nbsp;
+  <a href="https://twitter.com/kestra_io" style="margin: 0 10px;">
+        <img height="25" src="https://kestra.io/twitter.svg" alt="twitter" width="35" height="25" /></a>
+  <a href="https://www.linkedin.com/company/kestra/" style="margin: 0 10px;">
+        <img height="25" src="https://kestra.io/linkedin.svg" alt="linkedin" width="35" height="25" /></a> 
+  <a href="https://www.youtube.com/@kestra-io" style="margin: 0 10px;">
+        <img height="25" src="https://kestra.io/youtube.svg" alt="youtube" width="35" height="25" /></a>
 </p>
 
 <p align="center">
@@ -33,10 +36,10 @@
 
 <p align="center">
     <a href="https://go.kestra.io/video/product-overview" target="_blank">
-        <img src="https://kestra.io/startvideo.png" alt="Get started in 4 minutes with Kestra" width="640px" />
+        <img src="https://kestra.io/startvideo.png" alt="Get started in 3 minutes with Kestra" width="640px" />
     </a>
 </p>
-<p align="center" style="color:grey;"><i>Click on the image to learn how to get started with Kestra in 4 minutes.</i></p>
+<p align="center" style="color:grey;"><i>Click on the image to learn how to get started with Kestra in 3 minutes.</i></p>
 
 
 ## 🌟 What is Kestra?

build.gradle

@@ -25,7 +25,7 @@ plugins {
     id 'jacoco-report-aggregation'
 
     // helper
-    id "com.github.ben-manes.versions" version "0.52.0"
+    id "com.github.ben-manes.versions" version "0.53.0"
 
     // front
     id 'com.github.node-gradle.node' version '7.1.0'
@@ -37,7 +37,7 @@ plugins {
     id "com.vanniktech.maven.publish" version "0.34.0"
 
     // OWASP dependency check
-    id "org.owasp.dependencycheck" version "12.1.3" apply false
+    id "org.owasp.dependencycheck" version "12.1.5" apply false
 }
 
 idea {
@@ -168,8 +168,9 @@ allprojects {
 /**********************************************************************************************************************\
  * Test
  **********************************************************************************************************************/
-subprojects {
-    if (it.name != 'platform' && it.name != 'jmh-benchmarks') {
+subprojects {subProj ->
+
+    if (subProj.name != 'platform' && subProj.name != 'jmh-benchmarks') {
         apply plugin: "com.adarshr.test-logger"
 
         java {
@@ -207,6 +208,13 @@ subprojects {
 
         test {
             useJUnitPlatform()
+            reports {
+                junitXml.required = true
+                junitXml.outputPerTestCase = true
+                junitXml.mergeReruns = true
+                junitXml.includeSystemErrLog = true;
+                junitXml.outputLocation = layout.buildDirectory.dir("test-results/test")
+            }
 
             // set Xmx for test workers
             maxHeapSize = '4g'
@@ -222,6 +230,15 @@ subprojects {
             environment 'SECRET_PASSWORD', "cGFzc3dvcmQ="
             environment 'ENV_TEST1', "true"
             environment 'ENV_TEST2', "Pass by env"
+
+
+            if (subProj.name == 'core' || subProj.name == 'jdbc-h2' || subProj.name == 'jdbc-mysql' || subProj.name == 'jdbc-postgres') {
+                // JUnit 5 parallel settings
+                systemProperty 'junit.jupiter.execution.parallel.enabled', 'true'
+                systemProperty 'junit.jupiter.execution.parallel.mode.default', 'concurrent'
+                systemProperty 'junit.jupiter.execution.parallel.mode.classes.default', 'same_thread'
+                systemProperty 'junit.jupiter.execution.parallel.config.strategy', 'dynamic'
+            }
         }
 
         testlogger {

cli/build.gradle

@@ -40,5 +40,6 @@ dependencies {
     implementation project(":worker")
 
     //test
+    testImplementation project(':tests')
     testImplementation "org.wiremock:wiremock-jetty12"
 }

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

@@ -49,7 +49,7 @@ import java.util.concurrent.Callable;
 @Introspected
 public class App implements Callable<Integer> {
     public static void main(String[] args) {
-        execute(App.class, args);
+        execute(App.class, new String [] { Environment.CLI }, args);
     }
 
     @Override
@@ -57,13 +57,13 @@ public class App implements Callable<Integer> {
         return PicocliRunner.call(App.class, "--help");
     }
 
-    protected static void execute(Class<?> cls, String... args) {
+    protected static void execute(Class<?> cls, String[] environments, String... args) {
         // Log Bridge
         SLF4JBridgeHandler.removeHandlersForRootLogger();
         SLF4JBridgeHandler.install();
 
         // Init ApplicationContext
-        ApplicationContext applicationContext = App.applicationContext(cls, args);
+        ApplicationContext applicationContext = App.applicationContext(cls, environments, args);
 
         // Call Picocli command
         int exitCode = 0;
@@ -80,6 +80,7 @@ public class App implements Callable<Integer> {
         System.exit(Objects.requireNonNullElse(exitCode, 0));
     }
 
+
     /**
      * Create an {@link ApplicationContext} with additional properties based on configuration files (--config) and
      * forced Properties from current command.
@@ -88,12 +89,13 @@ public class App implements Callable<Integer> {
      * @return the application context created
      */
     protected static ApplicationContext applicationContext(Class<?> mainClass,
+                                                           String[] environments,
                                                            String[] args) {
 
         ApplicationContextBuilder builder = ApplicationContext
             .builder()
             .mainClass(mainClass)
-            .environments(Environment.CLI);
+            .environments(environments);
 
         CommandLine cmd = new CommandLine(mainClass, CommandLine.defaultFactory());
         continueOnParsingErrors(cmd);

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

@@ -2,19 +2,27 @@ package io.kestra.cli.commands.servers;
 
 import io.kestra.cli.AbstractCommand;
 import io.kestra.core.contexts.KestraContext;
-import jakarta.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
-abstract public class AbstractServerCommand extends AbstractCommand implements ServerCommandInterface {
+@Slf4j
+public abstract class AbstractServerCommand extends AbstractCommand implements ServerCommandInterface {
     @CommandLine.Option(names = {"--port"}, description = "The port to bind")
     Integer serverPort;
 
     @Override
     public Integer call()  throws Exception {
+        log.info("Machine information: {} available cpu(s), {}MB max memory, Java version {}", Runtime.getRuntime().availableProcessors(), maxMemoryInMB(), Runtime.version());
+
         this.shutdownHook(true, () -> KestraContext.getContext().shutdown());
+
         return super.call();
     }
 
+    private long maxMemoryInMB() {
+        return Runtime.getRuntime().maxMemory() / 1024 / 1024;
+    }
+
     protected static int defaultWorkerThread() {
         return Runtime.getRuntime().availableProcessors() * 8;
     }

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

@@ -3,7 +3,7 @@ package io.kestra.cli.commands.servers;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.ExecutorInterface;
-import io.kestra.executor.SkipExecutionService;
+import io.kestra.core.services.SkipExecutionService;
 import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;

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

@@ -4,10 +4,13 @@ import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
+import io.kestra.core.services.SkipExecutionService;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import picocli.CommandLine;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 @CommandLine.Command(
@@ -17,6 +20,11 @@ import java.util.Map;
 public class IndexerCommand extends AbstractServerCommand {
     @Inject
     private ApplicationContext applicationContext;
+    @Inject
+    private SkipExecutionService skipExecutionService;
+
+    @CommandLine.Option(names = {"--skip-indexer-records"}, split=",", description = "a list of indexer record keys, separated by a coma; for troubleshooting purpose only")
+    private List<String> skipIndexerRecords = Collections.emptyList();
 
     @SuppressWarnings("unused")
     public static Map<String, Object> propertiesOverrides() {
@@ -27,6 +35,8 @@ public class IndexerCommand extends AbstractServerCommand {
 
     @Override
     public Integer call() throws Exception {
+        this.skipExecutionService.setSkipIndexerRecords(skipIndexerRecords);
+
         super.call();
 
         Indexer indexer = applicationContext.getBean(Indexer.class);

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

@@ -7,7 +7,7 @@ import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
 import io.kestra.cli.StandAloneRunner;
-import io.kestra.executor.SkipExecutionService;
+import io.kestra.core.services.SkipExecutionService;
 import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
@@ -63,6 +63,9 @@ public class StandAloneCommand extends AbstractServerCommand {
     @CommandLine.Option(names = {"--skip-tenants"}, split=",", description = "a list of tenants to skip, separated by a coma; for troubleshooting purpose only")
     private List<String> skipTenants = Collections.emptyList();
 
+    @CommandLine.Option(names = {"--skip-indexer-records"}, split=",", description = "a list of indexer record keys, separated by a coma; for troubleshooting purpose only")
+    private List<String> skipIndexerRecords = Collections.emptyList();
+
     @CommandLine.Option(names = {"--no-tutorials"}, description = "Flag to disable auto-loading of tutorial flows.")
     boolean tutorialsDisabled = false;
 
@@ -93,6 +96,7 @@ public class StandAloneCommand extends AbstractServerCommand {
         this.skipExecutionService.setSkipFlows(skipFlows);
         this.skipExecutionService.setSkipNamespaces(skipNamespaces);
         this.skipExecutionService.setSkipTenants(skipTenants);
+        this.skipExecutionService.setSkipIndexerRecords(skipIndexerRecords);
         this.startExecutorService.applyOptions(startExecutors, notStartExecutors);
 
         KestraContext.getContext().injectWorkerConfigs(workerThread, null);

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

@@ -5,12 +5,15 @@ import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
+import io.kestra.core.services.SkipExecutionService;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 import picocli.CommandLine.Option;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
@@ -28,11 +31,17 @@ public class WebServerCommand extends AbstractServerCommand {
     @Inject
     private ExecutorsUtils executorsUtils;
 
+    @Inject
+    private SkipExecutionService skipExecutionService;
+
     @Option(names = {"--no-tutorials"}, description = "Flag to disable auto-loading of tutorial flows.")
-    boolean tutorialsDisabled = false;
+    private boolean tutorialsDisabled = false;
 
     @Option(names = {"--no-indexer"}, description = "Flag to disable starting an embedded indexer.")
-    boolean indexerDisabled = false;
+    private boolean indexerDisabled = false;
+
+    @CommandLine.Option(names = {"--skip-indexer-records"}, split=",", description = "a list of indexer record keys, separated by a coma; for troubleshooting purpose only")
+    private List<String> skipIndexerRecords = Collections.emptyList();
 
     @Override
     public boolean isFlowAutoLoadEnabled() {
@@ -48,6 +57,8 @@ public class WebServerCommand extends AbstractServerCommand {
 
     @Override
     public Integer call() throws Exception {
+        this.skipExecutionService.setSkipIndexerRecords(skipIndexerRecords);
+
         super.call();
 
         // start the indexer

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

@@ -262,6 +262,8 @@ public class FileChangedEventListener {
     }
 
     private String getTenantIdFromPath(Path path) {
+        // FIXME there is probably a bug here when a tenant has '_' in its name,
+        //  a valid tenant name is defined with following regex: "^[a-z0-9][a-z0-9_-]*"
         return path.getFileName().toString().split("_")[0];
     }
 }

cli/src/main/resources/application.yml

@@ -167,6 +167,9 @@ kestra:
       open-urls:
         - "/ping"
         - "/api/v1/executions/webhook/"
+        - "/api/v1/main/executions/webhook/"
+        - "/api/v1/*/executions/webhook/"
+        - "/api/v1/basicAuthValidationErrors"
 
     preview:
       initial-rows: 100

cli/src/test/java/io/kestra/cli/AppTest.java

@@ -37,7 +37,7 @@ class AppTest {
 
         final String[] args = new String[]{"server", serverType, "--help"};
 
-        try (ApplicationContext ctx = App.applicationContext(App.class, args)) {
+        try (ApplicationContext ctx = App.applicationContext(App.class, new String [] { Environment.CLI }, args)) {
             new CommandLine(App.class, new MicronautFactory(ctx)).execute(args);
 
             assertTrue(ctx.getProperty("kestra.server-type", ServerType.class).isEmpty());
@@ -52,7 +52,7 @@ class AppTest {
 
         final String[] argsWithMissingParams = new String[]{"flow", "namespace", "update"};
 
-        try (ApplicationContext ctx = App.applicationContext(App.class, argsWithMissingParams)) {
+        try (ApplicationContext ctx = App.applicationContext(App.class, new String [] { Environment.CLI }, argsWithMissingParams)) {
             new CommandLine(App.class, new MicronautFactory(ctx)).execute(argsWithMissingParams);
 
             assertThat(out.toString()).startsWith("Missing required parameters: ");

cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java

@@ -1,14 +1,15 @@
 package io.kestra.cli.services;
 
+import io.kestra.core.junit.annotations.FlakyTest;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.utils.Await;
+import io.kestra.core.utils.TestsUtils;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import org.apache.commons.io.FileUtils;
 import org.junit.jupiter.api.*;
-import org.junitpioneer.jupiter.RetryingTest;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -18,8 +19,8 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.junitpioneer.jupiter.RetryingTest;
 
-import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static io.kestra.core.utils.Rethrow.throwRunnable;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -57,10 +58,12 @@ class FileChangedEventListenerTest {
         }
     }
 
-    @RetryingTest(5) // Flaky on CI but always pass locally
+    @FlakyTest
+    @RetryingTest(2)
     void test() throws IOException, TimeoutException {
+        var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getSimpleName(), "test");
         // remove the flow if it already exists
-        flowRepository.findByIdWithSource(MAIN_TENANT, "io.kestra.tests.watch", "myflow").ifPresent(flow -> flowRepository.delete(flow));
+        flowRepository.findByIdWithSource(tenant, "io.kestra.tests.watch", "myflow").ifPresent(flow -> flowRepository.delete(flow));
 
         // create a basic flow
         String flow = """
@@ -73,14 +76,14 @@ class FileChangedEventListenerTest {
                 message: Hello World! 🚀
             """;
 
-        GenericFlow genericFlow = GenericFlow.fromYaml(MAIN_TENANT, flow);
+        GenericFlow genericFlow = GenericFlow.fromYaml(tenant, flow);
         Files.write(Path.of(FILE_WATCH + "/" + genericFlow.uidWithoutRevision() + ".yaml"), flow.getBytes());
         Await.until(
-            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "myflow").isPresent(),
+            () -> flowRepository.findById(tenant, "io.kestra.tests.watch", "myflow").isPresent(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );
-        Flow myflow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "myflow").orElseThrow();
+        Flow myflow = flowRepository.findById(tenant, "io.kestra.tests.watch", "myflow").orElseThrow();
         assertThat(myflow.getTasks()).hasSize(1);
         assertThat(myflow.getTasks().getFirst().getId()).isEqualTo("hello");
         assertThat(myflow.getTasks().getFirst().getType()).isEqualTo("io.kestra.plugin.core.log.Log");
@@ -88,16 +91,18 @@ class FileChangedEventListenerTest {
         // delete the flow
         Files.delete(Path.of(FILE_WATCH + "/" + genericFlow.uidWithoutRevision() + ".yaml"));
         Await.until(
-            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "myflow").isEmpty(),
+            () -> flowRepository.findById(tenant, "io.kestra.tests.watch", "myflow").isEmpty(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );
     }
 
-    @RetryingTest(5) // Flaky on CI but always pass locally
+    @FlakyTest
+    @RetryingTest(2)
     void testWithPluginDefault() throws IOException, TimeoutException {
+        var tenant = TestsUtils.randomTenant(FileChangedEventListenerTest.class.getName(), "testWithPluginDefault");
         // remove the flow if it already exists
-        flowRepository.findByIdWithSource(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").ifPresent(flow -> flowRepository.delete(flow));
+        flowRepository.findByIdWithSource(tenant, "io.kestra.tests.watch", "pluginDefault").ifPresent(flow -> flowRepository.delete(flow));
 
         // create a flow with plugin default
         String pluginDefault = """
@@ -113,14 +118,14 @@ class FileChangedEventListenerTest {
                 values:
                   message: Hello World!
             """;
-        GenericFlow genericFlow = GenericFlow.fromYaml(MAIN_TENANT, pluginDefault);
+        GenericFlow genericFlow = GenericFlow.fromYaml(tenant, pluginDefault);
         Files.write(Path.of(FILE_WATCH + "/" + genericFlow.uidWithoutRevision() + ".yaml"), pluginDefault.getBytes());
         Await.until(
-            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").isPresent(),
+            () -> flowRepository.findById(tenant, "io.kestra.tests.watch", "pluginDefault").isPresent(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );
-        Flow pluginDefaultFlow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").orElseThrow();
+        Flow pluginDefaultFlow = flowRepository.findById(tenant, "io.kestra.tests.watch", "pluginDefault").orElseThrow();
         assertThat(pluginDefaultFlow.getTasks()).hasSize(1);
         assertThat(pluginDefaultFlow.getTasks().getFirst().getId()).isEqualTo("helloWithDefault");
         assertThat(pluginDefaultFlow.getTasks().getFirst().getType()).isEqualTo("io.kestra.plugin.core.log.Log");
@@ -128,7 +133,7 @@ class FileChangedEventListenerTest {
         // delete both files
         Files.delete(Path.of(FILE_WATCH + "/" + genericFlow.uidWithoutRevision() + ".yaml"));
         Await.until(
-            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").isEmpty(),
+            () -> flowRepository.findById(tenant, "io.kestra.tests.watch", "pluginDefault").isEmpty(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );

core/build.gradle

@@ -84,7 +84,7 @@ dependencies {
 
     testImplementation "org.testcontainers:testcontainers:1.21.3"
     testImplementation "org.testcontainers:junit-jupiter:1.21.3"
-    testImplementation "org.bouncycastle:bcpkix-jdk18on:1.81"
+    testImplementation "org.bouncycastle:bcpkix-jdk18on"
 
     testImplementation "org.wiremock:wiremock-jetty12"
 }

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

@@ -118,7 +118,7 @@ public class JsonSchemaGenerator {
             removeRequiredOnPropsWithDefaults(objectNode);
 
             return MAPPER.convertValue(objectNode, MAP_TYPE_REFERENCE);
-        } catch (IllegalArgumentException e) {
+        } catch (Exception e) {
             throw new IllegalArgumentException("Unable to generate jsonschema for '" + cls.getName() + "'", e);
         }
     }

core/src/main/java/io/kestra/core/events/CrudEvent.java

@@ -3,30 +3,88 @@ package io.kestra.core.events;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.context.ServerRequestContext;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 
-@AllArgsConstructor
+import java.util.Objects;
+
 @Getter
 public class CrudEvent<T> {
-    T model;
+    private final T model;
     @Nullable
-    T previousModel;
-    CrudEventType type;
-    HttpRequest<?> request;
-
+    private final T previousModel;
+    private final CrudEventType type;
+    private final HttpRequest<?> request;
+    
+    /**
+     * Static helper method for creating a new {@link CrudEventType#UPDATE} CrudEvent.
+     *
+     * @param model the new created model.
+     * @param <T>   type of the model.
+     * @return the new {@link CrudEvent}.
+     */
+    public static <T> CrudEvent<T> create(T model) {
+        Objects.requireNonNull(model, "Can't create CREATE event with a null model");
+        return new CrudEvent<>(model, null, CrudEventType.CREATE);
+    }
+    
+    /**
+     * Static helper method for creating a new {@link CrudEventType#DELETE} CrudEvent.
+     *
+     * @param model the deleted model.
+     * @param <T>   type of the model.
+     * @return the new {@link CrudEvent}.
+     */
+    public static <T> CrudEvent<T> delete(T model) {
+        Objects.requireNonNull(model, "Can't create DELETE event with a null model");
+        return new CrudEvent<>(null, model, CrudEventType.DELETE);
+    }
+    
+    /**
+     * Static helper method for creating a new CrudEvent.
+     *
+     * @param before the model before the update.
+     * @param after  the model after the update.
+     * @param <T>   type of the model.
+     * @return the new {@link CrudEvent}.
+     */
+    public static <T> CrudEvent<T> of(T before, T after) {
+        
+        if (before == null && after == null) {
+            throw new IllegalArgumentException("Both before and after cannot be null");
+        }
+        
+        if (before == null) {
+            return create(after);
+        }
+        
+        if (after == null) {
+            return delete(before);
+        }
+        
+        return new CrudEvent<>(after, before, CrudEventType.UPDATE);
+    }
+    
+    /**
+     * @deprecated use the static factory methods.
+     */
+    @Deprecated
     public CrudEvent(T model, CrudEventType type) {
-        this.model = model;
-        this.type = type;
-        this.previousModel = null;
-        this.request = ServerRequestContext.currentRequest().orElse(null);
+        this(
+            CrudEventType.DELETE.equals(type) ? null : model,
+            CrudEventType.DELETE.equals(type) ? model : null, 
+            type, 
+            ServerRequestContext.currentRequest().orElse(null)
+        );
     }
 
     public CrudEvent(T model, T previousModel, CrudEventType type) {
+        this(model, previousModel, type, ServerRequestContext.currentRequest().orElse(null));
+    }
+    
+    public CrudEvent(T model, T previousModel, CrudEventType type, HttpRequest<?> request) {
         this.model = model;
         this.previousModel = previousModel;
         this.type = type;
-        this.request = ServerRequestContext.currentRequest().orElse(null);
+        this.request = request;
     }
-
 }

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

@@ -1,13 +1,16 @@
 package io.kestra.core.models;
 
 import io.kestra.core.utils.MapUtils;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.annotation.Nullable;
-import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.NotEmpty;
 
 import java.util.*;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
-public record Label(@NotNull String key, @NotNull String value) {
+@Schema(description = "A key/value pair that can be attached to a Flow or Execution. Labels are often used to organize and categorize objects.")
+public record Label(@NotEmpty String key, @NotEmpty String value) {
     public static final String SYSTEM_PREFIX = "system.";
 
     // system labels
@@ -41,7 +44,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     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)
+            .filter(label -> label.value() != null && !label.value().isEmpty() && label.key() != null && !label.key().isEmpty())
             // 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));
     }
@@ -56,6 +59,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     public static List<Label> deduplicate(@Nullable List<Label> labels) {
         if (labels == null || labels.isEmpty()) return Collections.emptyList();
         return toMap(labels).entrySet().stream()
+            .filter(getEntryNotEmptyPredicate())
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .collect(Collectors.toCollection(ArrayList::new));
     }
@@ -70,6 +74,7 @@ public record Label(@NotNull String key, @NotNull String value) {
         if (map == null || map.isEmpty()) return List.of();
         return map.entrySet()
             .stream()
+            .filter(getEntryNotEmptyPredicate())
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .toList();
     }
@@ -88,4 +93,14 @@ public record Label(@NotNull String key, @NotNull String value) {
         }
         return map;
     }
+
+    /**
+     * Provides predicate for not empty entries.
+     *
+     * @return The non-empty filter
+     */
+    public static Predicate<Map.Entry<String, String>> getEntryNotEmptyPredicate() {
+        return entry -> entry.getKey() != null && !entry.getKey().isEmpty() &&
+            entry.getValue() != null && !entry.getValue().isEmpty();
+    }
 }

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

@@ -1,16 +1,33 @@
 package io.kestra.core.models;
 
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.Valid;
 import jakarta.validation.constraints.Pattern;
 
+import java.util.List;
+import java.util.Map;
+
 /**
  * Interface that can be implemented by classes supporting plugin versioning.
  *
  * @see Plugin
  */
 public interface PluginVersioning {
+    
+    String TITLE = "Plugin Version";
+    String DESCRIPTION = """
+    Defines the version of the plugin to use.
 
-    @Pattern(regexp="\\d+\\.\\d+\\.\\d+(-[a-zA-Z0-9-]+)?|([a-zA-Z0-9]+)")
-    @Schema(title = "The version of the plugin to use.")
+    The version must follow the Semantic Versioning (SemVer) specification:
+      - A single-digit MAJOR version (e.g., `1`).
+      - A MAJOR.MINOR version (e.g., `1.1`).
+      - A MAJOR.MINOR.PATCH version, optionally with any qualifier
+        (e.g., `1.1.2`, `1.1.0-SNAPSHOT`).
+    """;
+    
+    @Schema(
+        title = TITLE,
+        description = DESCRIPTION
+    )
     String getVersion();
 }

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

@@ -254,19 +254,7 @@ public record QueryFilter(
          *
          * @return List of {@code ResourceField} with resource names, fields, and operations.
          */
-        public static List<ResourceField> asResourceList() {
-            return Arrays.stream(values())
-                .map(Resource::toResourceField)
-                .toList();
-        }
-
-        private static ResourceField toResourceField(Resource resource) {
-            List<FieldOp> fieldOps = resource.supportedField().stream()
-                .map(Resource::toFieldInfo)
-                .toList();
-            return new ResourceField(resource.name().toLowerCase(), fieldOps);
-        }
-
+        
         private static FieldOp toFieldInfo(Field field) {
             List<Operation> operations = field.supportedOp().stream()
                 .map(Resource::toOperation)
@@ -279,9 +267,6 @@ public record QueryFilter(
         }
     }
 
-    public record ResourceField(String name, List<FieldOp> fields) {
-    }
-
     public record FieldOp(String name, String value, List<Operation> operations) {
     }
 

core/src/main/java/io/kestra/core/models/collectors/ExecutionUsage.java

@@ -17,31 +17,12 @@ import java.util.List;
 @Introspected
 public class ExecutionUsage {
     private final List<DailyExecutionStatistics> dailyExecutionsCount;
-    private final List<DailyExecutionStatistics> dailyTaskRunsCount;
 
     public static ExecutionUsage of(final String tenantId,
                                     final ExecutionRepositoryInterface executionRepository,
                                     final ZonedDateTime from,
                                     final ZonedDateTime to) {
 
-        List<DailyExecutionStatistics> dailyTaskRunsCount = null;
-
-        try {
-            dailyTaskRunsCount = executionRepository.dailyStatistics(
-                null,
-                tenantId,
-                null,
-                null,
-                null,
-                from,
-                to,
-                DateUtils.GroupType.DAY,
-                null,
-                true);
-        } catch (UnsupportedOperationException ignored) {
-
-        }
-
         return ExecutionUsage.builder()
             .dailyExecutionsCount(executionRepository.dailyStatistics(
                 null,
@@ -52,28 +33,13 @@ public class ExecutionUsage {
                 from,
                 to,
                 DateUtils.GroupType.DAY,
-                null,
-                false))
-            .dailyTaskRunsCount(dailyTaskRunsCount)
+                null))
             .build();
     }
 
     public static ExecutionUsage of(final ExecutionRepositoryInterface repository,
                                     final ZonedDateTime from,
                                     final ZonedDateTime to) {
-        List<DailyExecutionStatistics> dailyTaskRunsCount = null;
-        try {
-            dailyTaskRunsCount = repository.dailyStatisticsForAllTenants(
-                null,
-                null,
-                null,
-                from,
-                to,
-                DateUtils.GroupType.DAY,
-                true
-            );
-        } catch (UnsupportedOperationException ignored) {}
-
         return ExecutionUsage.builder()
             .dailyExecutionsCount(repository.dailyStatisticsForAllTenants(
                 null,
@@ -81,10 +47,8 @@ public class ExecutionUsage {
                 null,
                 from,
                 to,
-                DateUtils.GroupType.DAY,
-                false
+                DateUtils.GroupType.DAY
             ))
-            .dailyTaskRunsCount(dailyTaskRunsCount)
             .build();
     }
 }

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

@@ -272,7 +272,7 @@ public class Execution implements DeletedInterface, TenantInterface {
     }
 
     public Execution withTaskRun(TaskRun taskRun) throws InternalException {
-        ArrayList<TaskRun> newTaskRunList = new ArrayList<>(this.taskRunList);
+        ArrayList<TaskRun> newTaskRunList = this.taskRunList == null ? new ArrayList<>() : new ArrayList<>(this.taskRunList);
 
         boolean b = Collections.replaceAll(
             newTaskRunList,
@@ -865,20 +865,18 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param e the exception raise
      * @return new taskRun with updated attempt with logs
      */
-    private FailedTaskRunWithLog lastAttemptsTaskRunForFailedExecution(TaskRun taskRun,
-        TaskRunAttempt lastAttempt, Exception e) {
+    private FailedTaskRunWithLog lastAttemptsTaskRunForFailedExecution(TaskRun taskRun, TaskRunAttempt lastAttempt, Exception e) {
+        TaskRun failed = taskRun
+            .withAttempts(
+                Stream
+                    .concat(
+                        taskRun.getAttempts().stream().limit(taskRun.getAttempts().size() - 1),
+                        Stream.of(lastAttempt.getState().isFailed() ? lastAttempt : lastAttempt.withState(State.Type.FAILED))
+                    )
+                    .toList()
+            );
         return new FailedTaskRunWithLog(
-            taskRun
-                .withAttempts(
-                    Stream
-                        .concat(
-                            taskRun.getAttempts().stream().limit(taskRun.getAttempts().size() - 1),
-                            Stream.of(lastAttempt
-                                .withState(State.Type.FAILED))
-                        )
-                        .toList()
-                )
-                .withState(State.Type.FAILED),
+            failed.getState().isFailed() ? failed : failed.withState(State.Type.FAILED),
             RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(taskRun, kind))
         );
     }

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

@@ -296,7 +296,7 @@ public class TaskRun implements TenantInterface {
     }
 
     public TaskRun incrementIteration() {
-        int iteration = this.iteration == null ? 1 : this.iteration;
+        int iteration = this.iteration == null ? 0 : this.iteration;
         return this.toBuilder()
             .iteration(iteration + 1)
             .build();

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

@@ -3,7 +3,6 @@ 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;
@@ -61,7 +60,14 @@ public abstract class AbstractFlow implements FlowInterface {
 
     @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
     @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
-    @Schema(implementation = Object.class, oneOf = {List.class, Map.class})
+    @Schema(
+            description = "Labels as a list of Label (key/value pairs) or as a map of string to string.",
+            oneOf = {
+                    Label[].class,
+                    Map.class
+            }
+    )
+    @Valid
     List<Label> labels;
 
     @Schema(additionalProperties = Schema.AdditionalPropertiesValue.TRUE)
@@ -69,4 +75,5 @@ public abstract class AbstractFlow implements FlowInterface {
 
     @Valid
     private WorkerGroup workerGroup;
+
 }

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

@@ -86,10 +86,11 @@ public class State {
 
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)
     public Duration getDuration() {
-        return Duration.between(
-            this.histories.getFirst().getDate(),
-            this.histories.size() > 1 ? this.histories.get(this.histories.size() - 1).getDate() : Instant.now()
-        );
+        if(this.getEndDate().isPresent()){
+            return Duration.between(this.getStartDate(), this.getEndDate().get());
+        } else {
+            return Duration.between(this.getStartDate(), Instant.now());
+        }
     }
 
     @JsonProperty(access = JsonProperty.Access.READ_ONLY)

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

@@ -30,7 +30,7 @@ public class ResolvedTask {
 
     public NextTaskRun toNextTaskRunIncrementIteration(Execution execution, Integer iteration) {
         return new NextTaskRun(
-            TaskRun.of(execution, this).withIteration(iteration != null ? iteration : 1),
+            TaskRun.of(execution, this).withIteration(iteration != null ? iteration : 0),
             this.getTask()
         );
     }

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

@@ -185,34 +185,6 @@ public class Trigger extends TriggerContext implements HasUID {
             .build();
     }
 
-    public static Trigger update(Trigger currentTrigger, Trigger newTrigger, ZonedDateTime nextExecutionDate) throws Exception {
-        Trigger updated = currentTrigger;
-
-        // If a backfill is created, we update the currentTrigger
-        // and set the nextExecutionDate() as the previous one
-        if (newTrigger.getBackfill() != null) {
-            updated = currentTrigger.toBuilder()
-                .backfill(
-                    newTrigger
-                        .getBackfill()
-                        .toBuilder()
-                        .end(newTrigger.getBackfill().getEnd() != null ? newTrigger.getBackfill().getEnd() : ZonedDateTime.now())
-                        .currentDate(
-                            newTrigger.getBackfill().getStart()
-                        )
-                        .previousNextExecutionDate(
-                            currentTrigger.getNextExecutionDate())
-                        .build())
-                .build();
-        }
-
-        return updated.toBuilder()
-            .nextExecutionDate(newTrigger.getDisabled() ?
-                null : nextExecutionDate)
-            .disabled(newTrigger.getDisabled())
-            .build();
-    }
-
     public Trigger resetExecution(Flow flow, Execution execution, ConditionContext conditionContext) {
         boolean disabled = this.getStopAfter() != null ? this.getStopAfter().contains(execution.getState().getCurrent()) : this.getDisabled();
         if (!disabled) {
@@ -276,27 +248,22 @@ public class Trigger extends TriggerContext implements HasUID {
             .build();
     }
 
-    public Trigger initBackfill(Trigger newTrigger) {
-        // If a backfill is created, we update the currentTrigger
+    public Trigger withBackfill(final Backfill backfill) {
+        Trigger updated = this;
+        // If a backfill is created, we update the trigger
         // and set the nextExecutionDate() as the previous one
-        if (newTrigger.getBackfill() != null) {
-
-            return this.toBuilder()
+        if (backfill != null) {
+            updated = this.toBuilder()
                 .backfill(
-                    newTrigger
-                        .getBackfill()
+                    backfill
                         .toBuilder()
-                        .end(newTrigger.getBackfill().getEnd() != null ? newTrigger.getBackfill().getEnd() : ZonedDateTime.now())
-                        .currentDate(
-                            newTrigger.getBackfill().getStart()
-                        )
-                        .previousNextExecutionDate(
-                            this.getNextExecutionDate())
+                        .end(backfill.getEnd() != null ? backfill.getEnd() : ZonedDateTime.now())
+                        .currentDate(backfill.getStart())
+                        .previousNextExecutionDate(this.getNextExecutionDate())
                         .build())
                 .build();
         }
-
-        return this;
+        return updated;
     }
 
     // if the next date is after the backfill end, we remove the backfill

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

@@ -56,7 +56,7 @@ public class DefaultPluginRegistry implements PluginRegistry {
      *
      * @return the {@link DefaultPluginRegistry}.
      */
-    public static DefaultPluginRegistry getOrCreate() {
+    public synchronized static DefaultPluginRegistry getOrCreate() {
         DefaultPluginRegistry instance = LazyHolder.INSTANCE;
         if (!instance.isInitialized()) {
             instance.init();
@@ -74,7 +74,7 @@ public class DefaultPluginRegistry implements PluginRegistry {
     /**
      * Initializes the registry by loading all core plugins.
      */
-    protected void init() {
+    protected synchronized void init() {
         if (initialized.compareAndSet(false, true)) {
             register(scanner.scan());
         }
@@ -200,7 +200,7 @@ public class DefaultPluginRegistry implements PluginRegistry {
         if (existing != null && existing.crc32() == plugin.crc32()) {
             return; // same plugin already registered
         }
-        
+
         lock.lock();
         try {
             if (existing != null) {
@@ -212,7 +212,7 @@ public class DefaultPluginRegistry implements PluginRegistry {
             lock.unlock();
         }
     }
-    
+
     protected void registerAll(Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> plugins) {
         pluginClassByIdentifier.putAll(plugins);
     }

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

@@ -6,6 +6,12 @@ import lombok.Getter;
 import lombok.ToString;
 
 import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Enumeration;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.zip.CRC32;
 
 @AllArgsConstructor
 @Getter
@@ -14,5 +20,59 @@ import java.net.URL;
 public class ExternalPlugin {
     private final URL location;
     private final URL[] resources;
-    private final long crc32;
+    private volatile Long crc32; // lazy-val
+    
+    public ExternalPlugin(URL location, URL[] resources) {
+        this.location = location;
+        this.resources = resources;
+    }
+    
+    public Long getCrc32() {
+        if (this.crc32 == null) {
+            synchronized (this) {
+                if (this.crc32 == null) {
+                    this.crc32 = computeJarCrc32(location);
+                }
+            }
+        }
+        return crc32;
+    }
+    
+    /**
+     * Compute a CRC32 of the JAR File without reading the whole file
+     *
+     * @param location of the JAR File.
+     * @return the CRC32 of {@code -1} if the checksum can't be computed.
+     */
+    private static long computeJarCrc32(final URL location) {
+        CRC32 crc = new CRC32();
+        try (JarFile jar = new JarFile(location.toURI().getPath(), false)) {
+            Enumeration<JarEntry> entries = jar.entries();
+            byte[] buffer = new byte[Long.BYTES]; // reusable buffer to avoid re-allocation
+            
+            while (entries.hasMoreElements()) {
+                JarEntry entry = entries.nextElement();
+                crc.update(entry.getName().getBytes(StandardCharsets.UTF_8));
+                updateCrc32WithLong(crc, buffer, entry.getSize());
+                updateCrc32WithLong(crc, buffer, entry.getCrc());
+            }
+            
+            return crc.getValue();
+        } catch (Exception e) {
+            return -1;
+        }
+    }
+    
+    private static void updateCrc32WithLong(CRC32 crc32, byte[] reusable, long val) {
+        // fast long -> byte conversion
+        reusable[0] = (byte) (val >>> 56);
+        reusable[1] = (byte) (val >>> 48);
+        reusable[2] = (byte) (val >>> 40);
+        reusable[3] = (byte) (val >>> 32);
+        reusable[4] = (byte) (val >>> 24);
+        reusable[5] = (byte) (val >>> 16);
+        reusable[6] = (byte) (val >>> 8);
+        reusable[7] = (byte) val;
+        crc32.update(reusable);;
+    }
 }

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

@@ -51,8 +51,7 @@ public class PluginResolver {
                 final List<URL> resources = resolveUrlsForPluginPath(path);
                 plugins.add(new ExternalPlugin(
                     path.toUri().toURL(),
-                    resources.toArray(new URL[0]),
-                    computeJarCrc32(path)
+                    resources.toArray(new URL[0])
                 ));
             }
         } catch (final InvalidPathException | MalformedURLException e) {
@@ -124,33 +123,5 @@ public class PluginResolver {
 
         return urls;
     }
-    
-    
-    /**
-     * Compute a CRC32 of the JAR File without reading the whole file
-     *
-     * @param location of the JAR File.
-     * @return the CRC32 of {@code -1} if the checksum can't be computed.
-     */
-    private static long computeJarCrc32(final Path location) {
-        CRC32 crc = new CRC32();
-        try (JarFile jar = new JarFile(location.toFile(), false)) {
-            Enumeration<JarEntry> entries = jar.entries();
-            while (entries.hasMoreElements()) {
-                JarEntry entry = entries.nextElement();
-                crc.update(entry.getName().getBytes());
-                crc.update(longToBytes(entry.getSize()));
-                crc.update(longToBytes(entry.getCrc()));
-            }
-        } catch (Exception e) {
-            return -1;
-        }
-        return crc.getValue();
-    }
-    
-    private static byte[] longToBytes(long x) {
-        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
-        buffer.putLong(x);
-        return buffer.array();
-    }
+
 }

core/src/main/java/io/kestra/core/plugins/serdes/PluginDeserializer.java

@@ -144,7 +144,7 @@ public final class PluginDeserializer<T extends Plugin> extends JsonDeserializer
 
     static String extractPluginRawIdentifier(final JsonNode node, final boolean isVersioningSupported) {
         String type = Optional.ofNullable(node.get(TYPE)).map(JsonNode::textValue).orElse(null);
-        String version = Optional.ofNullable(node.get(VERSION)).map(JsonNode::textValue).orElse(null);
+        String version = Optional.ofNullable(node.get(VERSION)).map(JsonNode::asText).orElse(null);
 
         if (type == null || type.isEmpty()) {
             return null;

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

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

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

@@ -25,8 +25,6 @@ import java.util.Optional;
 import java.util.function.Function;
 
 public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Execution>, QueryBuilderInterface<Executions.Fields> {
-    Boolean isTaskRunEnabled();
-
     default Optional<Execution> findById(String tenantId, String id) {
         return findById(tenantId, id, false);
     }
@@ -96,24 +94,19 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
 
     Flux<Execution> findAllAsync(@Nullable String tenantId);
 
-    ArrayListTotal<TaskRun> findTaskRun(
-        Pageable pageable,
-        @Nullable String tenantId,
-        List<QueryFilter> filters
-    );
-
     Execution delete(Execution execution);
 
     Integer purge(Execution execution);
 
+    Integer purge(List<Execution> executions);
+
     List<DailyExecutionStatistics> dailyStatisticsForAllTenants(
         @Nullable String query,
         @Nullable String namespace,
         @Nullable String flowId,
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
-        @Nullable DateUtils.GroupType groupBy,
-        boolean isTaskRun
+        @Nullable DateUtils.GroupType groupBy
     );
 
     List<DailyExecutionStatistics> dailyStatistics(
@@ -125,8 +118,7 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
         @Nullable DateUtils.GroupType groupBy,
-        List<State.Type> state,
-        boolean isTaskRun
+        List<State.Type> state
     );
 
     @Getter

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

@@ -4,6 +4,7 @@ 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.*;
+import io.kestra.plugin.core.dashboard.data.Flows;
 import io.micronaut.data.model.Pageable;
 import jakarta.annotation.Nullable;
 import jakarta.validation.ConstraintViolationException;
@@ -11,7 +12,7 @@ import jakarta.validation.ConstraintViolationException;
 import java.util.List;
 import java.util.Optional;
 
-public interface FlowRepositoryInterface {
+public interface FlowRepositoryInterface extends QueryBuilderInterface<Flows.Fields> {
 
     Optional<Flow> findById(String tenantId, String namespace, String id, Optional<Integer> revision, Boolean allowDeleted);
 
@@ -162,4 +163,6 @@ public interface FlowRepositoryInterface {
     FlowWithSource update(GenericFlow flow, FlowInterface previous) throws ConstraintViolationException;
 
     FlowWithSource delete(FlowInterface flow);
+
+    Boolean existAnyNoAcl(String tenantId);
 }

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

@@ -83,7 +83,9 @@ public class LocalFlowRepositoryLoader {
     }
 
     public void load(String tenantId, File basePath) throws IOException {
-        Map<String, FlowInterface> flowByUidInRepository = flowRepository.findAllForAllTenants().stream()
+        Map<String, FlowInterface> flowByUidInRepository = flowRepository.findAllForAllTenants()
+            .stream()
+            .filter(flow -> tenantId.equals(flow.getTenantId()))
             .collect(Collectors.toMap(FlowId::uidWithoutRevision, Function.identity()));
 
         try (Stream<Path> pathStream = Files.walk(basePath.toPath())) {

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

@@ -90,6 +90,8 @@ public interface LogRepositoryInterface extends SaveRepositoryInterface<LogEntry
 
     Integer purge(Execution execution);
 
+    Integer purge(List<Execution> executions);
+
     void deleteByQuery(String tenantId, String executionId, String taskId, String taskRunId, Level minLevel, Integer attempt);
 
     void deleteByQuery(String tenantId, String namespace, String flowId, String triggerId);

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

@@ -29,6 +29,8 @@ public interface MetricRepositoryInterface extends SaveRepositoryInterface<Metri
 
     Integer purge(Execution execution);
 
+    Integer purge(List<Execution> executions);
+
     Flux<MetricEntry> findAllAsync(@Nullable String tenantId);
 
     default Function<String, String> sortMapping() throws IllegalArgumentException {

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

@@ -5,10 +5,7 @@ import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.FlowInterface;
-import io.kestra.core.models.flows.FlowWithException;
-import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.*;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.ExecutableTask;
 import io.kestra.core.models.tasks.Task;
@@ -29,6 +26,7 @@ import org.apache.commons.lang3.stream.Streams;
 import java.time.Instant;
 import java.time.ZonedDateTime;
 import java.util.*;
+import java.util.stream.Collectors;
 
 import static io.kestra.core.trace.Tracer.throwCallable;
 import static io.kestra.core.utils.Rethrow.throwConsumer;
@@ -153,14 +151,22 @@ public final class ExecutableUtils {
                     currentFlow.getNamespace(),
                     currentFlow.getId()
                 )
-                .orElseThrow(() -> new IllegalStateException("Unable to find flow '" + subflowNamespace + "'.'" + subflowId + "' with revision '" + subflowRevision.orElse(0) + "'"));
+                .orElseThrow(() -> {
+                    String msg = "Unable to find flow '" + subflowNamespace + "'.'" + subflowId + "' with revision '" + subflowRevision.orElse(0) + "'";
+                    runContext.logger().error(msg);
+                    return new IllegalStateException(msg);
+                });
 
             if (flow.isDisabled()) {
-                throw new IllegalStateException("Cannot execute a flow which is disabled");
+                String msg = "Cannot execute a flow which is disabled";
+                runContext.logger().error(msg);
+                throw new IllegalStateException(msg);
             }
 
             if (flow instanceof FlowWithException fwe) {
-                throw new IllegalStateException("Cannot execute an invalid flow: " + fwe.getException());
+                String msg = "Cannot execute an invalid flow: " + fwe.getException();
+                runContext.logger().error(msg);
+                throw new IllegalStateException(msg);
             }
 
             List<Label> newLabels = inheritLabels ? new ArrayList<>(filterLabels(currentExecution.getLabels(), flow)) : new ArrayList<>(systemLabels(currentExecution));
@@ -201,7 +207,20 @@ public final class ExecutableUtils {
                     .build()
                 )
                 .withScheduleDate(scheduleOnDate);
-
+                if(execution.getInputs().size()<inputs.size()) {
+                    Map<String,Object>resolvedInputs=execution.getInputs();
+                    for (var inputKey : inputs.keySet()) {
+                        if (!resolvedInputs.containsKey(inputKey)) {
+                            runContext.logger().warn(
+                                "Input {} was provided by parent execution {} for subflow {}.{} but isn't declared at the subflow inputs",
+                                inputKey,
+                                currentExecution.getId(),
+                                currentTask.subflowId().namespace(),
+                                currentTask.subflowId().flowId()
+                            );
+                        }
+                    }
+                }
             // inject the traceparent into the new execution
             propagator.ifPresent(pg -> pg.inject(Context.current(), execution, ExecutionTextMapSetter.INSTANCE));
 

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

@@ -32,5 +32,7 @@ public class ExecutionRunning implements HasUID {
         return IdUtils.fromPartsAndSeparator('|', this.tenantId, this.namespace, this.flowId, this.execution.getId());
     }
 
-    public enum ConcurrencyState { CREATED, RUNNING, QUEUED, CANCELLED, FAILED }
+    // Note: the KILLED state is only used in the Kafka implementation to difference between purging terminated running execution (null)
+    // and purging killed execution which need special treatment
+    public enum ConcurrencyState { CREATED, RUNNING, QUEUED, CANCELLED, FAILED, KILLED }
 }

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

@@ -49,15 +49,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalTime;
-import java.util.AbstractMap;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.regex.Matcher;
@@ -231,6 +223,19 @@ public class FlowInputOutput {
                 return new AbstractMap.SimpleEntry<>(it.input().getId(), it.value());
             })
             .collect(HashMap::new, (m,v)-> m.put(v.getKey(), v.getValue()), HashMap::putAll);
+        if (resolved.size() < data.size()) {
+            RunContext runContext = runContextFactory.of(flow, execution);
+            for (var inputKey : data.keySet()) {
+                if (!resolved.containsKey(inputKey)) {
+                    runContext.logger().warn(
+                        "Input {} was provided for workflow {}.{} but isn't declared in the workflow inputs",
+                        inputKey,
+                        flow.getNamespace(),
+                        flow.getId()
+                    );
+                }
+            }
+        }
         return MapUtils.flattenToNestedMap(resolved);
     }
 
@@ -313,15 +318,15 @@ public class FlowInputOutput {
             });
             resolvable.setInput(input);
 
-            
+
             Object value = resolvable.get().value();
-            
+
             // resolve default if needed
             if (value == null && input.getDefaults() != null) {
                 value = resolveDefaultValue(input, runContext);
                 resolvable.isDefault(true);
             }
-            
+
             // validate and parse input value
             if (value == null) {
                 if (input.getRequired()) {
@@ -350,7 +355,7 @@ public class FlowInputOutput {
 
         return resolvable.get();
     }
-    
+
     public static Object resolveDefaultValue(Input<?> input, PropertyContext renderer) throws IllegalVariableEvaluationException {
         return switch (input.getType()) {
             case STRING, ENUM, SELECT, SECRET, EMAIL -> resolveDefaultPropertyAs(input, renderer, String.class);
@@ -367,7 +372,7 @@ public class FlowInputOutput {
             case MULTISELECT -> resolveDefaultPropertyAsList(input, renderer, String.class);
         };
     }
-    
+
     @SuppressWarnings("unchecked")
     private static <T> Object resolveDefaultPropertyAs(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
         return Property.as((Property<T>) input.getDefaults(), renderer, clazz);
@@ -376,7 +381,7 @@ public class FlowInputOutput {
     private static <T> Object resolveDefaultPropertyAsList(Input<?> input, PropertyContext renderer, Class<T> clazz) throws IllegalVariableEvaluationException {
         return Property.asList((Property<List<T>>) input.getDefaults(), renderer, clazz);
     }
-    
+
     private RunContext buildRunContextForExecutionAndInputs(final FlowInterface flow, final Execution execution, Map<String, InputAndValue> dependencies) {
         Map<String, Object> flattenInputs = MapUtils.flattenToNestedMap(dependencies.entrySet()
             .stream()
@@ -453,7 +458,7 @@ public class FlowInputOutput {
         if (data.getType() == null) {
             return Optional.of(new AbstractMap.SimpleEntry<>(data.getId(), current));
         }
-        
+
         final Type elementType = data instanceof ItemTypeInterface itemTypeInterface ? itemTypeInterface.getItemType() : null;
 
         return Optional.of(new AbstractMap.SimpleEntry<>(
@@ -530,17 +535,17 @@ public class FlowInputOutput {
             throw new Exception("Expected `" + type + "` but received `" + current + "` with errors:\n```\n" + e.getMessage() + "\n```");
         }
     }
-    
+
     public static Map<String, Object> renderFlowOutputs(List<Output> outputs, RunContext runContext) throws IllegalVariableEvaluationException {
         if (outputs == null) return Map.of();
-        
+
         // render required outputs
         Map<String, Object> outputsById = outputs
             .stream()
             .filter(output -> output.getRequired() == null || output.getRequired())
             .collect(HashMap::new, (map, entry) -> map.put(entry.getId(), entry.getValue()), Map::putAll);
         outputsById = runContext.render(outputsById);
-        
+
         // render optional outputs one by one to catch, log, and skip any error.
         for (io.kestra.core.models.flows.Output output : outputs) {
             if (Boolean.FALSE.equals(output.getRequired())) {
@@ -583,9 +588,9 @@ public class FlowInputOutput {
         }
 
         public void isDefault(boolean isDefault) {
-            this.input = new InputAndValue(this.input.input(), this.input.value(), this.input.enabled(), isDefault, this.input.exception());  
+            this.input = new InputAndValue(this.input.input(), this.input.value(), this.input.enabled(), isDefault, this.input.exception());
         }
-        
+
         public void setInput(final Input<?> input) {
             this.input = new InputAndValue(input, this.input.value(), this.input.enabled(), this.input.isDefault(), this.input.exception());
         }

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

@@ -500,7 +500,7 @@ public class FlowableUtils {
 
         ArrayList<ResolvedTask> result = new ArrayList<>();
 
-        int index = 0;
+        int iteration = 0;
         for (Object current : distinctValue) {
             try {
                 String resolvedValue = current instanceof String stringValue ? stringValue : MAPPER.writeValueAsString(current);
@@ -508,7 +508,7 @@ public class FlowableUtils {
                     result.add(ResolvedTask.builder()
                         .task(task)
                         .value(resolvedValue)
-                        .iteration(index++)
+                        .iteration(iteration)
                         .parentId(parentTaskRun.getId())
                         .build()
                     );
@@ -516,6 +516,7 @@ public class FlowableUtils {
             } catch (JsonProcessingException e) {
                 throw new IllegalVariableEvaluationException(e);
             }
+            iteration++;
         }
 
         return result;

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

@@ -0,0 +1,13 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.utils.IdUtils;
+
+public record MultipleConditionEvent(Flow flow, Execution execution) implements HasUID {
+    @Override
+    public String uid() {
+        return IdUtils.fromParts(flow.uidWithoutRevision(), execution.getId());
+    }
+}

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

@@ -45,7 +45,7 @@ final class Secret {
         for (var entry: data.entrySet()) {
             if (entry.getValue() instanceof Map map) {
                 // if some value are of type EncryptedString we decode them and replace the object
-                if (EncryptedString.TYPE.equalsIgnoreCase((String)map.get("type"))) {
+                if (map.get("type") instanceof String typeStr && EncryptedString.TYPE.equalsIgnoreCase(typeStr)) {
                     try {
                         String decoded = decrypt((String) map.get("value"));
                         decryptedMap.put(entry.getKey(), decoded);

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

@@ -168,6 +168,7 @@ public class Extension extends AbstractExtension {
         functions.put("randomPort", new RandomPortFunction());
         functions.put("fileExists", fileExistsFunction);
         functions.put("isFileEmpty", isFileEmptyFunction);
+        functions.put("nanoId", new NanoIDFunction());
         functions.put("tasksWithState", new TasksWithStateFunction());
         functions.put(HttpFunction.NAME, httpFunction);
         return functions;

core/src/main/java/io/kestra/core/runners/pebble/filters/TimestampMicroFilter.java

@@ -30,6 +30,6 @@ public class TimestampMicroFilter extends AbstractDate implements Filter {
         ZoneId zoneId = zoneId(timeZone);
         ZonedDateTime date = convert(input, zoneId, existingFormat);
 
-        return String.valueOf(TimeUnit.SECONDS.toNanos(date.toEpochSecond()) + TimeUnit.NANOSECONDS.toMicros(date.getNano()));
+        return String.valueOf(TimeUnit.SECONDS.toMicros(date.toEpochSecond()) + TimeUnit.NANOSECONDS.toMicros(date.getNano()));
     }
 }

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

@@ -0,0 +1,66 @@
+package io.kestra.core.runners.pebble.functions;
+
+import io.pebbletemplates.pebble.error.PebbleException;
+import io.pebbletemplates.pebble.extension.Function;
+import io.pebbletemplates.pebble.template.EvaluationContext;
+import io.pebbletemplates.pebble.template.PebbleTemplate;
+
+import java.security.SecureRandom;
+import java.util.List;
+import java.util.Map;
+
+public class NanoIDFunction implements Function {
+
+    private static final int DEFAULT_LENGTH = 21;
+    private static final char[] DEFAULT_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz-_".toCharArray();
+    private static final SecureRandom secureRandom = new SecureRandom();
+
+    private static final String LENGTH = "length";
+    private static final String ALPHABET = "alphabet";
+
+    private static final int MAX_LENGTH = 1000;
+
+    @Override
+    public Object execute(
+        Map<String, Object> args, PebbleTemplate self, EvaluationContext context, int lineNumber) {
+        int length = DEFAULT_LENGTH;
+        if (args.containsKey(LENGTH) && (args.get(LENGTH) instanceof Long)) {
+            length = parseLength(args, self, lineNumber);
+        }
+        char[] alphabet = DEFAULT_ALPHABET;
+        if (args.containsKey(ALPHABET) && (args.get(ALPHABET) instanceof String)) {
+            alphabet = ((String) args.get(ALPHABET)).toCharArray();
+        }
+        return createNanoID(length, alphabet);
+    }
+
+    private static int parseLength(Map<String, Object> args, PebbleTemplate self, int lineNumber) {
+        var value = (Long) args.get(LENGTH);
+        if(value > MAX_LENGTH) {
+            throw new PebbleException(
+                null,
+                "The 'nanoId()' function field 'length' must be lower than: " + MAX_LENGTH,
+                lineNumber,
+                self.getName());
+        }
+        return Math.toIntExact(value);
+    }
+
+    @Override
+    public List<String> getArgumentNames() {
+        return List.of(LENGTH,ALPHABET);
+    }
+
+    String createNanoID(int length, char[] alphabet){
+        final char[] data = new char[length];
+        final byte[] bytes = new byte[length];
+        final int mask = alphabet.length-1;
+        secureRandom.nextBytes(bytes);
+        for (int i = 0; i < length; ++i) {
+            data[i] = alphabet[bytes[i] & mask];
+        }
+        return String.valueOf(data);
+    }
+
+
+}

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

@@ -163,31 +163,28 @@ public final class JacksonMapper {
             .build();
     }
 
-    public static Pair<JsonNode, JsonNode> getBiDirectionalDiffs(Object previous, Object current)  {
-        JsonNode previousJson = MAPPER.valueToTree(previous);
-        JsonNode newJson = MAPPER.valueToTree(current);
+    public static Pair<JsonNode, JsonNode> getBiDirectionalDiffs(Object before, Object after)  {
+        JsonNode beforeNode = MAPPER.valueToTree(before);
+        JsonNode afterNode = MAPPER.valueToTree(after);
 
-        JsonNode patchPrevToNew = JsonDiff.asJson(previousJson, newJson);
-        JsonNode patchNewToPrev = JsonDiff.asJson(newJson, previousJson);
+        JsonNode patch = JsonDiff.asJson(beforeNode, afterNode);
+        JsonNode revert = JsonDiff.asJson(afterNode, beforeNode);
 
-        return Pair.of(patchPrevToNew, patchNewToPrev);
+        return Pair.of(patch, revert);
     }
-
-    public static String applyPatches(Object object, List<JsonNode> patches) throws JsonProcessingException {
+    
+    public static JsonNode applyPatchesOnJsonNode(JsonNode jsonObject, List<JsonNode> patches) {
         for (JsonNode patch : patches) {
             try {
                 // Required for ES
-                if (patch.findValue("value") == null) {
-                    ((ObjectNode) patch.get(0)).set("value", (JsonNode) null);
+                if (patch.findValue("value") == null && !patch.isEmpty()) {
+                    ((ObjectNode) patch.get(0)).set("value", null);
                 }
-                JsonNode current = MAPPER.valueToTree(object);
-                object = JsonPatch.fromJson(patch).apply(current);
+                jsonObject = JsonPatch.fromJson(patch).apply(jsonObject);
             } catch (IOException | JsonPatchException e) {
                 throw new RuntimeException(e);
             }
         }
-        return MAPPER.writeValueAsString(object);
+        return jsonObject;
     }
-
-
 }

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

@@ -56,8 +56,7 @@ import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static io.kestra.core.utils.Rethrow.throwFunction;
-import static io.kestra.core.utils.Rethrow.throwPredicate;
+import static io.kestra.core.utils.Rethrow.*;
 
 @Singleton
 @Slf4j
@@ -431,7 +430,8 @@ public class ExecutionService {
         @Nullable String flowId,
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
-        @Nullable List<State.Type> state
+        @Nullable List<State.Type> state,
+        int batchSize
     ) throws IOException {
         PurgeResult purgeResult = this.executionRepository
             .find(
@@ -448,24 +448,27 @@ public class ExecutionService {
                 null,
                 true
             )
-            .map(throwFunction(execution -> {
+            .buffer(batchSize)
+            .map(throwFunction(executions -> {
                 PurgeResult.PurgeResultBuilder<?, ?> builder = PurgeResult.builder();
 
                 if (purgeExecution) {
-                    builder.executionsCount(this.executionRepository.purge(execution));
+                    builder.executionsCount(this.executionRepository.purge(executions));
                 }
 
                 if (purgeLog) {
-                    builder.logsCount(this.logRepository.purge(execution));
+                    builder.logsCount(this.logRepository.purge(executions));
                 }
 
                 if (purgeMetric) {
-                    builder.metricsCount(this.metricRepository.purge(execution));
+                    builder.metricsCount(this.metricRepository.purge(executions));
                 }
 
                 if (purgeStorage) {
-                    URI uri = StorageContext.forExecution(execution).getExecutionStorageURI(StorageContext.KESTRA_SCHEME);
-                    builder.storagesCount(storageInterface.deleteByPrefix(execution.getTenantId(), execution.getNamespace(), uri).size());
+                    executions.forEach(throwConsumer(execution -> {
+                        URI uri = StorageContext.forExecution(execution).getExecutionStorageURI(StorageContext.KESTRA_SCHEME);
+                        builder.storagesCount(storageInterface.deleteByPrefix(execution.getTenantId(), execution.getNamespace(), uri).size());
+                    }));
                 }
 
                 return (PurgeResult) builder.build();
@@ -716,7 +719,8 @@ public class ExecutionService {
             newExecution = execution.withState(killingOrAfterKillState);
         }
 
-        eventPublisher.publishEvent(new CrudEvent<>(newExecution, execution, CrudEventType.UPDATE));
+        // Because this method is expected to be called by the Executor we can return the Execution
+        // immediately without publishing a CrudEvent like it's done on pause/resume method.
         return newExecution;
     }
     public Execution kill(Execution execution, FlowInterface flow) {

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

@@ -1,36 +1,45 @@
-package io.kestra.executor;
+package io.kestra.core.services;
 
 import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
+import jakarta.annotation.Nullable;
 import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 
 @Singleton
+@Slf4j
 public class SkipExecutionService {
     private volatile List<String> skipExecutions = Collections.emptyList();
     private volatile List<FlowId> skipFlows = Collections.emptyList();
     private volatile List<NamespaceId> skipNamespaces = Collections.emptyList();
     private volatile List<String> skipTenants = Collections.emptyList();
+    private volatile List<String> skipIndexerRecords = Collections.emptyList();
 
     public synchronized void setSkipExecutions(List<String> skipExecutions) {
         this.skipExecutions = skipExecutions == null ? Collections.emptyList() : skipExecutions;
     }
 
     public synchronized void setSkipFlows(List<String> skipFlows) {
-        this.skipFlows = skipFlows == null ? Collections.emptyList() : skipFlows.stream().map(FlowId::from).toList();
+        this.skipFlows = skipFlows == null ? Collections.emptyList() : skipFlows.stream().map(FlowId::from).filter(Objects::nonNull).toList();
     }
 
     public synchronized void setSkipNamespaces(List<String> skipNamespaces) {
-        this.skipNamespaces = skipNamespaces == null ? Collections.emptyList() : skipNamespaces.stream().map(NamespaceId::from).toList();
+        this.skipNamespaces = skipNamespaces == null ? Collections.emptyList() : skipNamespaces.stream().map(NamespaceId::from).filter(Objects::nonNull).toList();
     }
 
     public synchronized void setSkipTenants(List<String> skipTenants) {
         this.skipTenants = skipTenants == null ? Collections.emptyList() : skipTenants;
     }
 
+    public synchronized void setSkipIndexerRecords(List<String> skipIndexerRecords) {
+        this.skipIndexerRecords = skipIndexerRecords == null ? Collections.emptyList() : skipIndexerRecords;
+    }
+
     /**
      * Warning: this method didn't check the flow, so it must be used only when neither of the others can be used.
      */
@@ -46,6 +55,14 @@ public class SkipExecutionService {
         return skipExecution(taskRun.getTenantId(), taskRun.getNamespace(), taskRun.getFlowId(), taskRun.getExecutionId());
     }
 
+    /**
+     * Skip an indexer records based on its key.
+     * @param key the record key as computed by <code>QueueService.key(record)</code>, can be null
+     */
+    public boolean skipIndexerRecord(@Nullable String key) {
+        return key != null && skipIndexerRecords.contains(key);
+    }
+
     @VisibleForTesting
     boolean skipExecution(String tenant, String namespace, String flow, String executionId) {
         return (tenant != null && skipTenants.contains(tenant)) ||
@@ -59,19 +76,31 @@ public class SkipExecutionService {
     }
 
     record FlowId(String tenant, String namespace, String flow) {
-        static FlowId from(String flowId) {
+        static @Nullable FlowId from(String flowId) {
             String[] parts = SkipExecutionService.splitIdParts(flowId);
             if (parts.length == 3) {
                 return new FlowId(parts[0], parts[1], parts[2]);
+            } else if (parts.length == 2) {
+                return new FlowId(null, parts[0], parts[1]);
+            } else {
+                log.error("Invalid flow skip with values: '{}'", flowId);
             }
-            return new FlowId(null, parts[0], parts[1]);
+
+            return null;
         }
     };
 
     record NamespaceId(String tenant, String namespace) {
-        static NamespaceId from(String namespaceId) {
+        static @Nullable NamespaceId from(String namespaceId) {
             String[] parts = SkipExecutionService.splitIdParts(namespaceId);
-            return new NamespaceId(parts[0], parts[1]);
+
+            if (parts.length == 2) {
+                return new NamespaceId(parts[0], parts[1]);
+            } else {
+                log.error("Invalid namespace skip with values:'{}'", namespaceId);
+            }
+
+            return null;
         }
     };
 }

core/src/main/java/io/kestra/core/storages/kv/KVMetadata.java

@@ -1,5 +1,6 @@
 package io.kestra.core.storages.kv;
 
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
 import java.time.Duration;
@@ -9,6 +10,7 @@ import java.util.Map;
 import java.util.Optional;
 
 @Getter
+@EqualsAndHashCode
 public class KVMetadata {
     private String description;
     private Instant expirationDate;
@@ -17,14 +19,18 @@ public class KVMetadata {
         if (ttl != null && ttl.isNegative()) {
             throw new IllegalArgumentException("ttl cannot be negative");
         }
-
-
+        
         this.description = description;
         if (ttl != null) {
             this.expirationDate = Instant.now().plus(ttl);
         }
     }
-
+    
+    public KVMetadata(String description, Instant expirationDate) {
+        this.description = description;
+        this.expirationDate = expirationDate;
+    }
+    
     public KVMetadata(Map<String, String> metadata) {
         if (metadata == null) {
             return;
@@ -46,4 +52,9 @@ public class KVMetadata {
         }
         return map;
     }
+    
+    @Override
+    public String toString() {
+        return "[description=" + description + ", expirationDate=" + expirationDate + "]";
+    }
 }

core/src/main/java/io/kestra/core/storages/kv/KVStore.java

@@ -4,7 +4,9 @@ import io.kestra.core.exceptions.ResourceExpiredException;
 import io.kestra.core.storages.StorageContext;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.URI;
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 import java.util.regex.Pattern;
@@ -104,8 +106,33 @@ public interface KVStore {
     default boolean exists(String key) throws IOException {
         return list().stream().anyMatch(kvEntry -> kvEntry.key().equals(key));
     }
-
-
+    
+    /**
+     * Finds a KV entry with associated metadata for a given key.
+     *
+     * @param key   the KV entry key.
+     * @return an optional of {@link KVValueAndMetadata}.
+     * 
+     * @throws UncheckedIOException if an error occurred while executing the operation on the K/V store.
+     */
+    default Optional<KVValueAndMetadata> findMetadataAndValue(final String key) throws UncheckedIOException {
+        try {
+            return get(key).flatMap(entry ->
+                {
+                    try {
+                        return getValue(entry.key()).map(current -> new KVValueAndMetadata(new KVMetadata(entry.description(), entry.expirationDate()), current.value()));
+                    } catch (IOException e) {
+                        throw new UncheckedIOException(e);
+                    } catch (ResourceExpiredException e) {
+                        return Optional.empty();
+                    }
+                }
+            );
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+    
     Pattern KEY_VALIDATOR_PATTERN = Pattern.compile("[a-zA-Z0-9][a-zA-Z0-9._-]*");
 
     /**

core/src/main/java/io/kestra/core/topologies/FlowTopologyService.java

@@ -18,6 +18,7 @@ import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.FlowTopologyRepositoryInterface;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.MapUtils;
 import io.kestra.plugin.core.condition.*;
 import io.micronaut.core.annotation.Nullable;
 import jakarta.inject.Inject;
@@ -175,9 +176,6 @@ public class FlowTopologyService {
     protected boolean isTriggerChild(Flow parent, Flow child) {
         List<AbstractTrigger> triggers = ListUtils.emptyOnNull(child.getTriggers());
 
-        // simulated execution: we add a "simulated" label so conditions can know that the evaluation is for a simulated execution
-        Execution execution = Execution.newExecution(parent, (f, e) -> null, List.of(SIMULATED_EXECUTION), Optional.empty());
-
         // keep only flow trigger
         List<io.kestra.plugin.core.trigger.Flow> flowTriggers = triggers
             .stream()
@@ -189,13 +187,16 @@ public class FlowTopologyService {
             return false;
         }
 
+        // simulated execution: we add a "simulated" label so conditions can know that the evaluation is for a simulated execution
+        Execution execution = Execution.newExecution(parent, (f, e) -> null, List.of(SIMULATED_EXECUTION), Optional.empty());
+
         boolean conditionMatch =  flowTriggers
             .stream()
             .flatMap(flow -> ListUtils.emptyOnNull(flow.getConditions()).stream())
             .allMatch(condition -> validateCondition(condition, parent, execution));
 
         boolean preconditionMatch = flowTriggers.stream()
-            .anyMatch(flow -> flow.getPreconditions() == null || validateMultipleConditions(flow.getPreconditions().getConditions(), parent, execution));
+            .anyMatch(flow -> flow.getPreconditions() == null || validatePreconditions(flow.getPreconditions(), parent, execution));
 
         return conditionMatch && preconditionMatch;
     }
@@ -239,11 +240,24 @@ public class FlowTopologyService {
     }
 
     private boolean isMandatoryMultipleCondition(Condition condition) {
-        return Stream
-            .of(
-                Expression.class
-            )
-            .anyMatch(aClass -> condition.getClass().isAssignableFrom(aClass));
+        return condition.getClass().isAssignableFrom(Expression.class);
+    }
+
+    private boolean validatePreconditions(io.kestra.plugin.core.trigger.Flow.Preconditions preconditions, FlowInterface child, Execution execution) {
+        boolean  upstreamFlowMatched = MapUtils.emptyOnNull(preconditions.getUpstreamFlowsConditions())
+            .values()
+            .stream()
+            .filter(c -> !isFilterCondition(c))
+            .anyMatch(c -> validateCondition(c, child, execution));
+
+        boolean  whereMatched = MapUtils.emptyOnNull(preconditions.getWhereConditions())
+            .values()
+            .stream()
+            .filter(c -> !isFilterCondition(c))
+            .allMatch(c -> validateCondition(c, child, execution));
+
+        // to be a dependency, if upstream flow is set it must be either inside it so it's a AND between upstream flow and where
+        return upstreamFlowMatched && whereMatched;
     }
 
     private boolean isFilterCondition(Condition condition) {

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

@@ -206,22 +206,17 @@ public class MapUtils {
 
     /**
      * Utility method that flatten a nested map.
-     * <p>
-     * NOTE: for simplicity, this method didn't allow to flatten maps with conflicting keys that would end up in different flatten keys,
-     * this could be related later if needed by flattening {k1: k2: {k3: v1}, k1: {k4: v2}} to {k1.k2.k3: v1, k1.k4: v2} is prohibited for now.
      *
      * @param nestedMap the nested map.
      * @return the flattened map.
-     *
-     * @throws IllegalArgumentException if any entry contains a map of more than one element.
      */
     public static Map<String, Object> nestedToFlattenMap(@NotNull Map<String, Object> nestedMap) {
         Map<String, Object> result = new TreeMap<>();
 
         for (Map.Entry<String, Object> entry : nestedMap.entrySet()) {
             if (entry.getValue() instanceof Map<?, ?> map) {
-                Map.Entry<String, Object> flatten = flattenEntry(entry.getKey(), (Map<String, Object>) map);
-                result.put(flatten.getKey(), flatten.getValue());
+                Map<String, Object> flatten = flattenEntry(entry.getKey(), (Map<String, Object>) map);
+                result.putAll(flatten);
             } else {
                 result.put(entry.getKey(), entry.getValue());
             }
@@ -229,18 +224,19 @@ public class MapUtils {
         return result;
     }
 
-    private static Map.Entry<String, Object> flattenEntry(String key, Map<String, Object> value) {
-        if (value.size() > 1) {
-            throw new IllegalArgumentException("You cannot flatten a map with an entry that is a map of more than one element, conflicting key: " + key);
+    private static Map<String, Object> flattenEntry(String key, Map<String, Object> value) {
+        Map<String, Object> result = new TreeMap<>();
+
+        for (Map.Entry<String, Object> entry : value.entrySet()) {
+            String newKey = key + "." + entry.getKey();
+            Object newValue = entry.getValue();
+            if (newValue instanceof Map<?, ?> map) {
+                result.putAll(flattenEntry(newKey, (Map<String, Object>) map));
+            } else {
+                result.put(newKey, newValue);
+            }
         }
 
-        Map.Entry<String, Object> entry = value.entrySet().iterator().next();
-        String newKey = key + "." + entry.getKey();
-        Object newValue = entry.getValue();
-        if (newValue instanceof Map<?, ?> map) {
-            return flattenEntry(newKey, (Map<String, Object>) map);
-        } else {
-            return Map.entry(newKey, newValue);
-        }
+        return result;
     }
 }

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

@@ -32,48 +32,84 @@ public class Version implements Comparable<Version> {
      * @param version   the version.
      * @return          a new {@link Version} instance.
      */
-    public static Version of(String version) {
+    public static Version of(final Object version) {
 
-        if (version.startsWith("v")) {
-            version = version.substring(1);
+        if (Objects.isNull(version)) {
+            throw new IllegalArgumentException("Invalid version, cannot parse null version");
+        }
+        
+        String strVersion = version.toString();
+        
+        if (strVersion.startsWith("v")) {
+            strVersion = strVersion.substring(1);
         }
 
-        int qualifier = version.indexOf("-");
+        int qualifier = strVersion.indexOf("-");
 
         final String[] versions = qualifier > 0 ?
-            version.substring(0, qualifier).split("\\.") :
-            version.split("\\.");
+            strVersion.substring(0, qualifier).split("\\.") :
+            strVersion.split("\\.");
         try {
             final int majorVersion = Integer.parseInt(versions[0]);
-            final int minorVersion = versions.length > 1 ? Integer.parseInt(versions[1]) : 0;
-            final int incrementalVersion = versions.length > 2 ? Integer.parseInt(versions[2]) : 0;
+            final Integer minorVersion = versions.length > 1 ? Integer.parseInt(versions[1]) : null;
+            final Integer incrementalVersion = versions.length > 2 ? Integer.parseInt(versions[2]) : null;
 
             return new Version(
                 majorVersion,
                 minorVersion,
                 incrementalVersion,
-                qualifier > 0 ? version.substring(qualifier + 1) : null,
-                version
+                qualifier > 0 ? strVersion.substring(qualifier + 1) : null,
+                strVersion
             );
         } catch (NumberFormatException e) {
             throw new IllegalArgumentException("Invalid version, cannot parse '" + version + "'");
         }
     }
-
+    
     /**
-     * Static helper method for returning the most recent stable version for a current {@link Version}.
+     * Resolves the most appropriate stable version from a collection, based on a given input version.
+     * <p>
+     * The matching rules are:
+     * <ul>
+     *   <li>If {@code from} specifies only a major version (e.g. {@code 1}), return the latest stable version
+     *       with the same major (e.g. {@code 1.2.3}).</li>
+     *   <li>If {@code from} specifies a major and minor version only (e.g. {@code 1.2}), return the latest
+     *       stable version with the same major and minor (e.g. {@code 1.2.3}).</li>
+     *   <li>If {@code from} specifies a full version with major, minor, and patch (e.g. {@code 1.2.2}),
+     *       then only return it if it is exactly present (and stable) in {@code versions}.
+     *       No "upgrade" is performed in this case.</li>
+     *   <li>If no suitable version is found, returns {@code null}.</li>
+     * </ul>
      *
-     * @param from     the current version.
-     * @param versions the list of version.
-     *
-     * @return the last stable version.
+     * @param from     the reference version (may specify only major, or major+minor, or major+minor+patch).
+     * @param versions the collection of candidate versions to resolve against.
+     * @return the best matching stable version, or {@code null} if none match.
      */
     public static Version getStable(final Version from, final Collection<Version> versions) {
-        List<Version> compatibleVersions = versions.stream()
-            .filter(v -> v.majorVersion() == from.majorVersion() && v.minorVersion() == from.minorVersion())
-            .toList();
-        if (compatibleVersions.isEmpty()) return null;
-        return Version.getLatest(compatibleVersions);
+        // Case 1: "from" is only a major (e.g. 1)
+        if (from.hasOnlyMajor()) {
+            List<Version> sameMajor = versions.stream()
+                .filter(v -> v.majorVersion() == from.majorVersion())
+                .toList();
+            return sameMajor.isEmpty() ? null : Version.getLatest(sameMajor);
+        }
+        
+        // Case 2: "from" is major+minor only (e.g. 1.2)
+        if (from.hasMajorAndMinorOnly()) {
+            List<Version> sameMinor = versions.stream()
+                .filter(v -> v.majorVersion() == from.majorVersion()
+                    && v.minorVersion() == from.minorVersion())
+                .toList();
+            return sameMinor.isEmpty() ? null : Version.getLatest(sameMinor);
+        }
+        
+        // Case 3: "from" is full version (major+minor+patch)
+        if (versions.contains(from)) {
+            return from;
+        }
+        
+        // No match
+        return null;
     }
 
     /**
@@ -123,8 +159,8 @@ public class Version implements Comparable<Version> {
     }
 
     private final int majorVersion;
-    private final int minorVersion;
-    private final int incrementalVersion;
+    private final Integer minorVersion;
+    private final Integer patchVersion;
     private final Qualifier qualifier;
 
     private final String originalVersion;
@@ -134,14 +170,14 @@ public class Version implements Comparable<Version> {
      *
      * @param majorVersion          the major version (must be superior or equal to 0).
      * @param minorVersion          the minor version (must be superior or equal to 0).
-     * @param incrementalVersion    the incremental version (must be superior or equal to 0).
+     * @param patchVersion    the incremental version (must be superior or equal to 0).
      * @param qualifier             the qualifier.
      */
     public Version(final int majorVersion,
                    final int minorVersion,
-                   final int incrementalVersion,
+                   final int patchVersion,
                    final String qualifier) {
-        this(majorVersion, minorVersion, incrementalVersion, qualifier, null);
+        this(majorVersion, minorVersion, patchVersion, qualifier, null);
     }
 
     /**
@@ -149,25 +185,25 @@ public class Version implements Comparable<Version> {
      *
      * @param majorVersion          the major version (must be superior or equal to 0).
      * @param minorVersion          the minor version (must be superior or equal to 0).
-     * @param incrementalVersion    the incremental version (must be superior or equal to 0).
+     * @param patchVersion    the incremental version (must be superior or equal to 0).
      * @param qualifier             the qualifier.
      * @param originalVersion       the original string version.
      */
-    private Version(final int majorVersion,
-                    final int minorVersion,
-                    final int incrementalVersion,
+    private Version(final Integer majorVersion,
+                    final Integer minorVersion,
+                    final Integer patchVersion,
                     final String qualifier,
                     final String originalVersion) {
         this.majorVersion =  requirePositive(majorVersion, "major");
         this.minorVersion = requirePositive(minorVersion, "minor");
-        this.incrementalVersion = requirePositive(incrementalVersion, "incremental");
+        this.patchVersion = requirePositive(patchVersion, "incremental");
         this.qualifier = qualifier != null ? new Qualifier(qualifier) : null;
         this.originalVersion = originalVersion;
     }
 
 
-    private static int requirePositive(int version, final String message) {
-        if (version < 0) {
+    private static Integer requirePositive(Integer version, final String message) {
+        if (version != null && version < 0) {
             throw new IllegalArgumentException(String.format("The '%s' version must super or equal to 0", message));
         }
         return version;
@@ -178,11 +214,11 @@ public class Version implements Comparable<Version> {
     }
 
     public int minorVersion() {
-        return minorVersion;
+        return minorVersion != null ? minorVersion : 0;
     }
 
-    public int incrementalVersion() {
-        return incrementalVersion;
+    public int patchVersion() {
+        return patchVersion != null ? patchVersion : 0;
     }
 
     public Qualifier qualifier() {
@@ -197,9 +233,9 @@ public class Version implements Comparable<Version> {
         if (this == o) return true;
         if (!(o instanceof Version)) return false;
         Version version = (Version) o;
-        return majorVersion == version.majorVersion &&
-            minorVersion == version.minorVersion &&
-            incrementalVersion == version.incrementalVersion &&
+        return Objects.equals(majorVersion,version.majorVersion) &&
+            Objects.equals(minorVersion, version.minorVersion) &&
+            Objects.equals(patchVersion,version.patchVersion) &&
             Objects.equals(qualifier, version.qualifier);
     }
 
@@ -208,7 +244,7 @@ public class Version implements Comparable<Version> {
      */
     @Override
     public int hashCode() {
-        return Objects.hash(majorVersion, minorVersion, incrementalVersion, qualifier);
+        return Objects.hash(majorVersion, minorVersion, patchVersion, qualifier);
     }
 
     /**
@@ -218,7 +254,7 @@ public class Version implements Comparable<Version> {
     public String toString() {
         if (originalVersion != null) return originalVersion;
 
-        String version =  majorVersion + "." + minorVersion + "." + incrementalVersion;
+        String version =  majorVersion + "." + minorVersion + "." + patchVersion;
         return (qualifier != null) ? version +"-" + qualifier : version;
     }
 
@@ -238,7 +274,7 @@ public class Version implements Comparable<Version> {
             return compareMinor;
         }
 
-        int compareIncremental = Integer.compare(that.incrementalVersion, this.incrementalVersion);
+        int compareIncremental = Integer.compare(that.patchVersion, this.patchVersion);
         if (compareIncremental != 0) {
             return compareIncremental;
         }
@@ -253,6 +289,21 @@ public class Version implements Comparable<Version> {
 
         return this.qualifier.compareTo(that.qualifier);
     }
+    
+    
+    /**
+     * @return true if only major is specified (e.g. "1")
+     */
+    private boolean hasOnlyMajor() {
+        return minorVersion == null && patchVersion == null;
+    }
+    
+    /**
+     * @return true if major+minor are specified, but no patch (e.g. "1.2")
+     */
+    private boolean hasMajorAndMinorOnly() {
+        return minorVersion != null && patchVersion == null;
+    }
 
     /**
      * Checks whether this version is before the given one.

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

@@ -46,16 +46,19 @@ public class VersionProvider {
         this.date = loadTime(gitProperties);
         this.version = loadVersion(buildProperties, gitProperties);
 
-        // check the version in the settings and update if needed, we did't use it would allow us to detect incompatible update later if needed
-        if (settingRepository.isPresent()) {
-            Optional<Setting> versionSetting = settingRepository.get().findByKey(Setting.INSTANCE_VERSION);
-            if (versionSetting.isEmpty() || !versionSetting.get().getValue().equals(this.version)) {
-                settingRepository.get().save(Setting.builder()
-                    .key(Setting.INSTANCE_VERSION)
-                    .value(this.version)
-                    .build()
-                );
-            }
+        // check the version in the settings and update if needed, we didn't use it would allow us to detect incompatible update later if needed
+        settingRepository.ifPresent(
+            settingRepositoryInterface -> persistVersion(settingRepositoryInterface, version));
+    }
+
+    private static synchronized void persistVersion(SettingRepositoryInterface settingRepositoryInterface, String version) {
+        Optional<Setting> versionSetting = settingRepositoryInterface.findByKey(Setting.INSTANCE_VERSION);
+        if (versionSetting.isEmpty() || !versionSetting.get().getValue().equals(version)) {
+            settingRepositoryInterface.save(Setting.builder()
+                .key(Setting.INSTANCE_VERSION)
+                .value(version)
+                .build()
+            );
         }
     }
 

core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java

@@ -63,7 +63,7 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
 
         List<String> violations = new ArrayList<>();
 
-        if (RESERVED_FLOW_IDS.contains(value.getId())) {
+        if (value.getId() != null && RESERVED_FLOW_IDS.contains(value.getId())) {
             violations.add("Flow id is a reserved keyword: " + value.getId() + ". List of reserved keywords: " + String.join(", ", RESERVED_FLOW_IDS));
         }
 

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

@@ -0,0 +1,52 @@
+package io.kestra.plugin.core.dashboard.data;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.dashboards.ColumnDescriptor;
+import io.kestra.core.models.dashboards.DataFilter;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.QueryBuilderInterface;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+@EqualsAndHashCode
+@Schema(
+    title = "Display Flow data in a dashboard chart."
+)
+@Plugin(
+    examples = {
+        @Example(
+            title = "Display a chart with a list of Flows.",
+            full = true,
+            code = { """
+            charts:
+              - id: list_flows
+                type: io.kestra.plugin.core.dashboard.chart.Table
+                data:
+                  type: io.kestra.plugin.core.dashboard.data.Flows
+                  columns:
+                    namespace:
+                      field: NAMESPACE
+                    id:
+                      field: ID
+            """
+          }
+        )
+    }
+)
+@JsonTypeName("Flows")
+public class Flows<C extends ColumnDescriptor<Flows.Fields>> extends DataFilter<Flows.Fields, C> implements IFlows {
+    @Override
+    public Class<? extends QueryBuilderInterface<Fields>> repositoryClass() {
+        return FlowRepositoryInterface.class;
+    }
+}

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

@@ -0,0 +1,51 @@
+package io.kestra.plugin.core.dashboard.data;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.dashboards.ColumnDescriptor;
+import io.kestra.core.models.dashboards.DataFilterKPI;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.QueryBuilderInterface;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+@EqualsAndHashCode
+@Schema(
+    title = "Display a chart with Flows KPI.",
+    description = "Change."
+)
+@Plugin(
+    examples = {
+        @Example(
+            title = "Display count of Flows.",
+            full = true,
+            code = { """
+            charts:
+              - id: kpi
+                    type: io.kestra.plugin.core.dashboard.chart.KPI
+                    data:
+                      type: io.kestra.plugin.core.dashboard.data.FlowsKPI
+                      columns:
+                        field: ID
+                        agg: COUNT
+            """
+          }
+        )
+    }
+)
+@JsonTypeName("FlowsKPI")
+public class FlowsKPI<C extends ColumnDescriptor<FlowsKPI.Fields>> extends DataFilterKPI<FlowsKPI.Fields, C> implements IFlows {
+    @Override
+    public Class<? extends QueryBuilderInterface<Fields>> repositoryClass() {
+        return FlowRepositoryInterface.class;
+    }
+}

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

@@ -0,0 +1,37 @@
+package io.kestra.plugin.core.dashboard.data;
+
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.models.dashboards.filters.AbstractFilter;
+import io.kestra.core.utils.ListUtils;
+
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+public interface IFlows extends IData<IFlows.Fields> {
+
+    default List<AbstractFilter<IFlows.Fields>> whereWithGlobalFilters(List<QueryFilter> filters, ZonedDateTime startDate, ZonedDateTime endDate, List<AbstractFilter<IFlows.Fields>> where) {
+        List<AbstractFilter<IFlows.Fields>> updatedWhere = where != null ? new ArrayList<>(where) : new ArrayList<>();
+
+        if (ListUtils.isEmpty(filters)) {
+            return updatedWhere;
+        }
+
+        List<QueryFilter> namespaceFilters = filters.stream().filter(f -> f.field().equals(QueryFilter.Field.NAMESPACE)).toList();
+        if (!namespaceFilters.isEmpty()) {
+            updatedWhere.removeIf(filter -> filter.getField().equals(IFlows.Fields.NAMESPACE));
+            namespaceFilters.forEach(f -> {
+                updatedWhere.add(f.toDashboardFilterBuilder(IFlows.Fields.NAMESPACE, f.value()));
+            });
+        }
+
+
+        return updatedWhere;
+    }
+
+    enum Fields {
+        ID,
+        NAMESPACE,
+        REVISION
+    }
+}

core/src/main/java/io/kestra/plugin/core/execution/Exit.java

@@ -21,6 +21,7 @@ import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.Optional;
 
@@ -68,6 +69,7 @@ import java.util.Optional;
         )
     }
 )
+@Slf4j
 public class Exit extends Task implements ExecutionUpdatableTask {
     @NotNull
     @Schema(
@@ -104,12 +106,13 @@ public class Exit extends Task implements ExecutionUpdatableTask {
                     // ends all parents
                     while (newTaskRun.getParentTaskRunId() != null) {
                         newTaskRun = newExecution.findTaskRunByTaskRunId(newTaskRun.getParentTaskRunId()).withState(exitState);
-                        newExecution = execution.withTaskRun(newTaskRun);
+                        newExecution = newExecution.withTaskRun(newTaskRun);
                     }
                     return newExecution;
                 } catch (InternalException e) {
                     // in case we cannot update the last not terminated task run, we ignore it
-                    return execution;
+                    log.warn("Unable to update the taskrun state", e);
+                    return execution.withState(exitState);
                 }
             })
             .orElse(execution)

core/src/main/java/io/kestra/plugin/core/execution/Labels.java

@@ -127,9 +127,24 @@ public class Labels extends Task implements ExecutionUpdatableTask {
         }
 
         // check for system labels: none can be passed at runtime
-        Optional<Map.Entry<String, String>> first = labelsAsMap.entrySet().stream().filter(entry -> entry.getKey().startsWith(SYSTEM_PREFIX)).findFirst();
-        if (first.isPresent()) {
-            throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().getKey() + "=" + first.get().getValue());
+        Optional<Map.Entry<String, String>> systemLabel = labelsAsMap.entrySet().stream()
+            .filter(entry -> entry.getKey().startsWith(SYSTEM_PREFIX))
+            .findFirst();
+        if (systemLabel.isPresent()) {
+            throw new IllegalArgumentException(
+                "System labels can only be set by Kestra itself, offending label: " +
+                systemLabel.get().getKey() + "=" + systemLabel.get().getValue()
+            );
+        }
+
+        // check for empty label values
+        Optional<Map.Entry<String, String>> emptyValue = labelsAsMap.entrySet().stream()
+            .filter(entry -> entry.getValue().isEmpty())
+            .findFirst();
+        if (emptyValue.isPresent()) {
+            throw new IllegalArgumentException(
+                "Label values cannot be empty, offending label: " + emptyValue.get().getKey()
+            );
         }
 
         Map<String, String> newLabels = ListUtils.emptyOnNull(execution.getLabels()).stream()
@@ -140,6 +155,7 @@ public class Labels extends Task implements ExecutionUpdatableTask {
         newLabels.putAll(labelsAsMap);
 
         return execution.withLabels(newLabels.entrySet().stream()
+            .filter(Label.getEntryNotEmptyPredicate())
             .map(entry -> new Label(
                 entry.getKey(),
                 entry.getValue()

core/src/main/java/io/kestra/plugin/core/execution/PurgeExecutions.java

@@ -102,6 +102,14 @@ public class PurgeExecutions extends Task implements RunnableTask<PurgeExecution
     @Builder.Default
     private Property<Boolean> purgeStorage = Property.ofValue(true);
 
+    @Schema(
+        title = "The size of the bulk delete",
+        description = "For performance, deletion is made by batch of by default 100 executions/logs/metrics."
+    )
+    @Builder.Default
+    @NotNull
+    private Property<Integer> batchSize = Property.ofValue(100);
+
     @Override
     public PurgeExecutions.Output run(RunContext runContext) throws Exception {
         ExecutionService executionService = ((DefaultRunContext)runContext).getApplicationContext().getBean(ExecutionService.class);
@@ -124,9 +132,10 @@ public class PurgeExecutions extends Task implements RunnableTask<PurgeExecution
             flowInfo.tenantId(),
             renderedNamespace,
             runContext.render(flowId).as(String.class).orElse(null),
-            startDate != null ? ZonedDateTime.parse(runContext.render(startDate).as(String.class).orElseThrow()) : null,
+            runContext.render(startDate).as(String.class).map(ZonedDateTime::parse).orElse(null),
             ZonedDateTime.parse(runContext.render(endDate).as(String.class).orElseThrow()),
-            this.states == null ? null : runContext.render(this.states).asList(State.Type.class)
+            this.states == null ? null : runContext.render(this.states).asList(State.Type.class),
+            runContext.render(this.batchSize).as(Integer.class).orElseThrow()
         );
 
         return Output.builder()

core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java

@@ -478,7 +478,7 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
             try (InputStream is = runContext.storage().getFile(splitsURI)){
                 String fileContent = new String(is.readAllBytes());
                 List<URI> splits = fileContent.lines().map(line -> URI.create(line)).toList();
-                AtomicInteger currentIteration = new AtomicInteger(1);
+                AtomicInteger currentIteration = new AtomicInteger(0);
 
                 return splits
                     .stream()

core/src/main/java/io/kestra/plugin/core/flow/Subflow.java

@@ -216,49 +216,46 @@ public class Subflow extends Task implements ExecutableTask<Subflow.Output>, Chi
 
         VariablesService variablesService = ((DefaultRunContext) runContext).getApplicationContext().getBean(VariablesService.class);
         if (this.wait) { // we only compute outputs if we wait for the subflow
-            boolean isOutputsAllowed = runContext
-                .<Boolean>pluginConfiguration(PLUGIN_FLOW_OUTPUTS_ENABLED)
-                .orElse(true);
-
             List<io.kestra.core.models.flows.Output> subflowOutputs = flow.getOutputs();
-            
+
             // region [deprecated] Subflow outputs feature
-            if (subflowOutputs == null && isOutputsAllowed && this.getOutputs() != null) {
-                subflowOutputs = this.getOutputs().entrySet().stream()
-                    .<io.kestra.core.models.flows.Output>map(entry -> io.kestra.core.models.flows.Output
-                        .builder()
-                        .id(entry.getKey())
-                        .value(entry.getValue())
-                        .required(true)
-                        .build()
-                    )
-                    .toList();
+            if (subflowOutputs == null && this.getOutputs() != null) {
+                boolean isOutputsAllowed = runContext
+                    .<Boolean>pluginConfiguration(PLUGIN_FLOW_OUTPUTS_ENABLED)
+                    .orElse(true);
+                if (isOutputsAllowed) {
+                    try {
+                        subflowOutputs = this.getOutputs().entrySet().stream()
+                            .<io.kestra.core.models.flows.Output>map(entry -> io.kestra.core.models.flows.Output
+                                .builder()
+                                .id(entry.getKey())
+                                .value(entry.getValue())
+                                .required(true)
+                                .build()
+                            )
+                            .toList();
+                    } catch (Exception e) {
+                        Variables variables = variablesService.of(StorageContext.forTask(taskRun), builder.build());
+                        return failSubflowDueToOutput(runContext, taskRun, execution, e, variables);
+                    }
+                } else {
+                    runContext.logger().warn("Defining outputs inside the Subflow task is not allowed.");
+                }
             }
             //endregion
 
             if (subflowOutputs != null && !subflowOutputs.isEmpty()) {
                 try {
-                    Map<String, Object> outputs = FlowInputOutput.renderFlowOutputs(subflowOutputs, runContext);
-                    
+                    Map<String, Object> rOutputs = FlowInputOutput.renderFlowOutputs(subflowOutputs, runContext);
+
                     FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class); // this is hacking
                     if (flow.getOutputs() != null && flowInputOutput != null) {
-                        outputs = flowInputOutput.typedOutputs(flow, execution, outputs);
+                        rOutputs = flowInputOutput.typedOutputs(flow, execution, rOutputs);
                     }
-                    builder.outputs(outputs);
+                    builder.outputs(rOutputs);
                 } catch (Exception e) {
-                    runContext.logger().warn("Failed to extract outputs with the error: '{}'", e.getLocalizedMessage(), e);
-                    var state = State.Type.fail(this);
                     Variables variables = variablesService.of(StorageContext.forTask(taskRun), builder.build());
-                    taskRun = taskRun
-                        .withState(state)
-                        .withAttempts(Collections.singletonList(TaskRunAttempt.builder().state(new State().withState(state)).build()))
-                        .withOutputs(variables);
-
-                    return Optional.of(SubflowExecutionResult.builder()
-                        .executionId(execution.getId())
-                        .state(State.Type.FAILED)
-                        .parentTaskRun(taskRun)
-                        .build());
+                    return failSubflowDueToOutput(runContext, taskRun, execution, e, variables);
                 }
             }
         }
@@ -282,6 +279,21 @@ public class Subflow extends Task implements ExecutableTask<Subflow.Output>, Chi
         return Optional.of(ExecutableUtils.subflowExecutionResult(taskRun, execution));
     }
 
+    private Optional<SubflowExecutionResult> failSubflowDueToOutput(RunContext runContext, TaskRun taskRun, Execution execution, Exception e, Variables outputs) {
+        runContext.logger().error("Failed to extract outputs with the error: '{}'", e.getLocalizedMessage(), e);
+        var state = State.Type.fail(this);
+        taskRun = taskRun
+            .withState(state)
+            .withAttempts(Collections.singletonList(TaskRunAttempt.builder().state(new State().withState(state)).build()))
+            .withOutputs(outputs);
+
+        return Optional.of(SubflowExecutionResult.builder()
+            .executionId(execution.getId())
+            .state(State.Type.FAILED)
+            .parentTaskRun(taskRun)
+            .build());
+    }
+
     @Override
     public boolean waitForExecution() {
         return this.wait;

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

@@ -202,7 +202,7 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
             code = """
                 id: sentry_execution_example
                 namespace: company.team
-                
+
                 tasks:
                 - id: send_alert
                   type: io.kestra.plugin.notifications.sentry.SentryExecution
@@ -221,7 +221,7 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
                         - WARNING
                     - type: io.kestra.plugin.core.condition.ExecutionNamespace
                         namespace: company.payroll
-                        prefix: false"""                                
+                        prefix: false"""
         )
 
     },
@@ -405,6 +405,28 @@ public class Flow extends AbstractTrigger implements TriggerOutput<Flow.Output>
             return conditions;
         }
 
+        @JsonIgnore
+        public Map<String, Condition> getUpstreamFlowsConditions() {
+            AtomicInteger conditionId = new AtomicInteger();
+            return ListUtils.emptyOnNull(flows).stream()
+                .map(upstreamFlow -> Map.entry(
+                    "condition_" + conditionId.incrementAndGet(),
+                    new UpstreamFlowCondition(upstreamFlow)
+                ))
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        }
+
+        @JsonIgnore
+        public Map<String, Condition> getWhereConditions() {
+            AtomicInteger conditionId = new AtomicInteger();
+            return ListUtils.emptyOnNull(where).stream()
+                .map(filter -> Map.entry(
+                    "condition_" + conditionId.incrementAndGet() + "_" + filter.getId(),
+                    new FilterCondition(filter)
+                ))
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        }
+
         @Override
         public Logger logger() {
             return log;

core/src/test/java/io/kestra/core/docs/DocumentationGeneratorTest.java

@@ -21,10 +21,13 @@ import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 @KestraTest
+@Execution(ExecutionMode.SAME_THREAD)
 class DocumentationGeneratorTest {
     @Inject
     JsonSchemaGenerator jsonSchemaGenerator;

core/src/test/java/io/kestra/core/events/CrudEventTest.java

@@ -0,0 +1,121 @@
+package io.kestra.core.events;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+class CrudEventTest {
+    
+    @Test
+    void shouldReturnCreateEventWhenModelIsProvided() {
+        // Given
+        String model = "testModel";
+        
+        // When
+        CrudEvent<String> event = CrudEvent.create(model);
+        
+        // Then
+        assertThat(event.getModel()).isEqualTo(model);
+        assertThat(event.getPreviousModel()).isNull();
+        assertThat(event.getType()).isEqualTo(CrudEventType.CREATE);
+        assertThat(event.getRequest()).isNull();
+    }
+    
+    @Test
+    void shouldThrowExceptionWhenCreateEventWithNullModel() {
+        // Given
+        String model = null;
+        
+        // When / Then
+        assertThatThrownBy(() -> CrudEvent.create(model))
+            .isInstanceOf(NullPointerException.class)
+            .hasMessage("Can't create CREATE event with a null model");
+    }
+    
+    @Test
+    void shouldReturnDeleteEventWhenModelIsProvided() {
+        // Given
+        String model = "testModel";
+        
+        // When
+        CrudEvent<String> event = CrudEvent.delete(model);
+        
+        // Then
+        assertThat(event.getModel()).isNull();
+        assertThat(event.getPreviousModel()).isEqualTo(model);
+        assertThat(event.getType()).isEqualTo(CrudEventType.DELETE);
+        assertThat(event.getRequest()).isNull();
+    }
+    
+    @Test
+    void shouldThrowExceptionWhenDeleteEventWithNullModel() {
+        // Given
+        String model = null;
+        
+        // When / Then
+        assertThatThrownBy(() -> CrudEvent.delete(model))
+            .isInstanceOf(NullPointerException.class)
+            .hasMessage("Can't create DELETE event with a null model");
+    }
+    
+    @Test
+    void shouldReturnUpdateEventWhenBeforeAndAfterAreProvided() {
+        // Given
+        String before = "oldModel";
+        String after = "newModel";
+        
+        // When
+        CrudEvent<String> event = CrudEvent.of(before, after);
+        
+        // Then
+        assertThat(event.getModel()).isEqualTo(after);
+        assertThat(event.getPreviousModel()).isEqualTo(before);
+        assertThat(event.getType()).isEqualTo(CrudEventType.UPDATE);
+        assertThat(event.getRequest()).isNull();
+    }
+    
+    @Test
+    void shouldReturnCreateEventWhenBeforeIsNullAndAfterIsProvided() {
+        // Given
+        String before = null;
+        String after = "newModel";
+        
+        // When
+        CrudEvent<String> event = CrudEvent.of(before, after);
+        
+        // Then
+        assertThat(event.getModel()).isEqualTo(after);
+        assertThat(event.getPreviousModel()).isNull();
+        assertThat(event.getType()).isEqualTo(CrudEventType.CREATE);
+        assertThat(event.getRequest()).isNull();
+    }
+    
+    @Test
+    void shouldReturnDeleteEventWhenAfterIsNullAndBeforeIsProvided() {
+        // Given
+        String before = "oldModel";
+        String after = null;
+        
+        // When
+        CrudEvent<String> event = CrudEvent.of(before, after);
+        
+        // Then
+        assertThat(event.getModel()).isNull();
+        assertThat(event.getPreviousModel()).isEqualTo(before);
+        assertThat(event.getType()).isEqualTo(CrudEventType.DELETE);
+        assertThat(event.getRequest()).isNull();
+    }
+    
+    @Test
+    void shouldThrowExceptionWhenBothBeforeAndAfterAreNull() {
+        // Given
+        String before = null;
+        String after = null;
+        
+        // When / Then
+        assertThatThrownBy(() -> CrudEvent.of(before, after))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessage("Both before and after cannot be null");
+    }
+}

core/src/test/java/io/kestra/core/http/client/HttpClientTest.java

@@ -37,6 +37,7 @@ import lombok.Value;
 import org.apache.commons.io.IOUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -67,6 +68,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 
 @KestraTest
 @Testcontainers
+@org.junit.jupiter.api.parallel.Execution(ExecutionMode.SAME_THREAD)
 class HttpClientTest {
     @Inject
     private ApplicationContext applicationContext;

core/src/test/java/io/kestra/core/models/LabelTest.java

@@ -1,19 +1,32 @@
 package io.kestra.core.models;
 
+import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.models.validations.ModelValidator;
+import jakarta.inject.Inject;
+import jakarta.validation.ConstraintViolationException;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+@KestraTest
 class LabelTest {
+    @Inject
+    private ModelValidator modelValidator;
 
     @Test
     void shouldGetNestedMapGivenDistinctLabels() {
         Map<String, Object> result = Label.toNestedMap(List.of(
             new Label(Label.USERNAME, "test"),
-            new Label(Label.CORRELATION_ID, "id"))
+            new Label(Label.CORRELATION_ID, "id"),
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null)
+            )
         );
 
         assertThat(result).isEqualTo(
@@ -34,6 +47,18 @@ class LabelTest {
         );
     }
 
+    @Test
+    void toNestedMapShouldIgnoreEmptyOrNull() {
+        Map<String, Object> result = Label.toNestedMap(List.of(
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null))
+        );
+
+        assertThat(result).isEmpty();
+    }
+
     @Test
     void shouldGetMapGivenDistinctLabels() {
         Map<String, String> result = Label.toMap(List.of(
@@ -59,6 +84,18 @@ class LabelTest {
         );
     }
 
+    @Test
+    void toMapShouldIgnoreEmptyOrNull() {
+        Map<String, String> result = Label.toMap(List.of(
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null))
+        );
+
+        assertThat(result).isEmpty();
+    }
+
     @Test
     void shouldDuplicateLabelsWithKeyOrderKept() {
         List<Label> result = Label.deduplicate(List.of(
@@ -73,4 +110,28 @@ class LabelTest {
             new Label(Label.CORRELATION_ID, "id")
         );
     }
+
+    @Test
+    void deduplicateShouldIgnoreEmptyAndNull() {
+        List<Label> result = Label.deduplicate(List.of(
+            new Label("", "bar"),
+            new Label(null, "bar"),
+            new Label("foo", ""),
+            new Label("baz", null))
+        );
+
+        assertThat(result).isEmpty();
+    }
+
+    @Test
+    void shouldValidateEmpty() {
+        Optional<ConstraintViolationException> validLabelResult = modelValidator.isValid(new Label("foo", "bar"));
+        assertThat(validLabelResult.isPresent()).isFalse();
+
+        Optional<ConstraintViolationException> emptyValueLabelResult = modelValidator.isValid(new Label("foo", ""));
+        assertThat(emptyValueLabelResult.isPresent()).isTrue();
+
+        Optional<ConstraintViolationException> emptyKeyLabelResult = modelValidator.isValid(new Label("", "bar"));
+        assertThat(emptyKeyLabelResult.isPresent()).isTrue();
+    }
 }

core/src/test/java/io/kestra/core/models/executions/ExecutionTest.java

@@ -13,19 +13,19 @@ import java.util.Map;
 import static org.assertj.core.api.Assertions.assertThat;
 
 class ExecutionTest {
-    private static final TaskRun.TaskRunBuilder TASK_RUN = TaskRun.builder()
-        .id("test");
 
     @Test
     void hasTaskRunJoinableTrue() {
         Execution execution = Execution.builder()
-            .taskRunList(Collections.singletonList(TASK_RUN
+            .taskRunList(Collections.singletonList(TaskRun.builder()
+                .id("test")
                 .state(new State(State.Type.RUNNING, new State()))
                 .build())
             )
             .build();
 
-        assertThat(execution.hasTaskRunJoinable(TASK_RUN
+        assertThat(execution.hasTaskRunJoinable(TaskRun.builder()
+            .id("test")
             .state(new State(State.Type.FAILED, new State()
                 .withState(State.Type.RUNNING)
             ))
@@ -36,13 +36,15 @@ class ExecutionTest {
     @Test
     void hasTaskRunJoinableSameState() {
         Execution execution = Execution.builder()
-            .taskRunList(Collections.singletonList(TASK_RUN
+            .taskRunList(Collections.singletonList(TaskRun.builder()
+                .id("test")
                 .state(new State())
                 .build())
             )
             .build();
 
-        assertThat(execution.hasTaskRunJoinable(TASK_RUN
+        assertThat(execution.hasTaskRunJoinable(TaskRun.builder()
+            .id("test")
             .state(new State())
             .build()
         )).isFalse();
@@ -51,7 +53,8 @@ class ExecutionTest {
     @Test
     void hasTaskRunJoinableFailedExecutionFromExecutor() {
         Execution execution = Execution.builder()
-            .taskRunList(Collections.singletonList(TASK_RUN
+            .taskRunList(Collections.singletonList(TaskRun.builder()
+                .id("test")
                 .state(new State(State.Type.FAILED, new State()
                     .withState(State.Type.RUNNING)
                 ))
@@ -59,7 +62,8 @@ class ExecutionTest {
             )
             .build();
 
-        assertThat(execution.hasTaskRunJoinable(TASK_RUN
+        assertThat(execution.hasTaskRunJoinable(TaskRun.builder()
+            .id("test")
             .state(new State(State.Type.RUNNING, new State()))
             .build()
         )).isFalse();
@@ -68,7 +72,8 @@ class ExecutionTest {
     @Test
     void hasTaskRunJoinableRestartFailed() {
         Execution execution = Execution.builder()
-            .taskRunList(Collections.singletonList(TASK_RUN
+            .taskRunList(Collections.singletonList(TaskRun.builder()
+                .id("test")
                 .state(new State(State.Type.CREATED, new State()
                     .withState(State.Type.RUNNING)
                     .withState(State.Type.FAILED)
@@ -77,7 +82,8 @@ class ExecutionTest {
             )
             .build();
 
-        assertThat(execution.hasTaskRunJoinable(TASK_RUN
+        assertThat(execution.hasTaskRunJoinable(TaskRun.builder()
+            .id("test")
             .state(new State(State.Type.FAILED, new State()
                 .withState(State.Type.RUNNING)
             ))
@@ -88,7 +94,8 @@ class ExecutionTest {
     @Test
     void hasTaskRunJoinableRestartSuccess() {
         Execution execution = Execution.builder()
-            .taskRunList(Collections.singletonList(TASK_RUN
+            .taskRunList(Collections.singletonList(TaskRun.builder()
+                .id("test")
                 .state(new State(State.Type.CREATED, new State()
                     .withState(State.Type.RUNNING)
                     .withState(State.Type.SUCCESS)
@@ -97,7 +104,8 @@ class ExecutionTest {
             )
             .build();
 
-        assertThat(execution.hasTaskRunJoinable(TASK_RUN
+        assertThat(execution.hasTaskRunJoinable(TaskRun.builder()
+            .id("test")
             .state(new State(State.Type.SUCCESS, new State()
                 .withState(State.Type.RUNNING)
                 .withState(State.Type.SUCCESS)
@@ -109,7 +117,8 @@ class ExecutionTest {
     @Test
     void hasTaskRunJoinableAfterRestart() {
         Execution execution = Execution.builder()
-            .taskRunList(Collections.singletonList(TASK_RUN
+            .taskRunList(Collections.singletonList(TaskRun.builder()
+                .id("test")
                 .state(new State(State.Type.CREATED, new State()
                     .withState(State.Type.RUNNING)
                     .withState(State.Type.FAILED)
@@ -118,7 +127,8 @@ class ExecutionTest {
             )
             .build();
 
-        assertThat(execution.hasTaskRunJoinable(TASK_RUN
+        assertThat(execution.hasTaskRunJoinable(TaskRun.builder()
+            .id("test")
             .state(new State(State.Type.SUCCESS, new State()
                 .withState(State.Type.RUNNING)
                 .withState(State.Type.FAILED)

core/src/test/java/io/kestra/core/models/executions/StateDurationTest.java

@@ -0,0 +1,55 @@
+package io.kestra.core.models.executions;
+
+import io.kestra.core.models.flows.State;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class StateDurationTest {
+
+
+    private static final Instant NOW = Instant.now();
+    private static final Instant ONE = NOW.minus(Duration.ofDays(1000));
+    private static final Instant TWO = ONE.plus(Duration.ofHours(11));
+    private static final Instant THREE = TWO.plus(Duration.ofHours(222));
+
+    @Test
+    void justCreated() {
+        var state = State.of(
+            State.Type.CREATED,
+            List.of(
+                new State.History(State.Type.CREATED, ONE)
+            )
+        );
+        assertThat(state.getDuration()).isCloseTo(Duration.between(ONE, NOW), Duration.ofMinutes(10));
+    }
+
+    @Test
+    void success() {
+        var state = State.of(
+            State.Type.SUCCESS,
+            List.of(
+                new State.History(State.Type.CREATED, ONE),
+                new State.History(State.Type.RUNNING, TWO),
+                new State.History(State.Type.SUCCESS, THREE)
+            )
+        );
+        assertThat(state.getDuration()).isEqualTo(Duration.between(ONE, THREE));
+    }
+
+    @Test
+    void isRunning() {
+        var state = State.of(
+            State.Type.RUNNING,
+            List.of(
+                new State.History(State.Type.CREATED, ONE),
+                new State.History(State.Type.RUNNING, TWO)
+            )
+        );
+        assertThat(state.getDuration()).isCloseTo(Duration.between(ONE, NOW), Duration.ofMinutes(10));
+    }
+}

core/src/test/java/io/kestra/core/models/hierarchies/FlowGraphTest.java

@@ -7,12 +7,11 @@ import io.kestra.core.junit.annotations.ExecuteFlow;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.junit.annotations.LoadFlows;
 import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.repositories.TriggerRepositoryInterface;
-import io.kestra.core.runners.RunnerUtils;
+import io.kestra.core.runners.TestRunnerUtils;
 import io.kestra.core.serializers.YamlParser;
 import io.kestra.core.services.GraphService;
 import io.kestra.core.utils.GraphUtils;
@@ -45,7 +44,7 @@ class FlowGraphTest {
     private TriggerRepositoryInterface triggerRepositoryInterface;
 
     @Inject
-    private RunnerUtils runnerUtils;
+    private TestRunnerUtils runnerUtils;
 
     @Test
     void simple() throws IllegalVariableEvaluationException, IOException {
@@ -261,10 +260,10 @@ class FlowGraphTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/task-flow.yaml",
-        "flows/valids/switch.yaml"})
+    @LoadFlows(value = {"flows/valids/task-flow.yaml",
+        "flows/valids/switch.yaml"}, tenantId = "tenant1")
     void subflow() throws IllegalVariableEvaluationException, IOException, FlowProcessingException {
-        FlowWithSource flow = this.parse("flows/valids/task-flow.yaml");
+        FlowWithSource flow = this.parse("flows/valids/task-flow.yaml", "tenant1");
         FlowGraph flowGraph = GraphUtils.flowGraph(flow, null);
 
         assertThat(flowGraph.getNodes().size()).isEqualTo(6);
@@ -293,15 +292,15 @@ class FlowGraphTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/task-flow-dynamic.yaml",
-        "flows/valids/switch.yaml"})
+    @LoadFlows(value = {"flows/valids/task-flow-dynamic.yaml",
+        "flows/valids/switch.yaml"}, tenantId = "tenant2")
     void dynamicIdSubflow() throws IllegalVariableEvaluationException, TimeoutException, QueueException, IOException, FlowProcessingException {
-        FlowWithSource flow = this.parse("flows/valids/task-flow-dynamic.yaml").toBuilder().revision(1).build();
+        FlowWithSource flow = this.parse("flows/valids/task-flow-dynamic.yaml", "tenant2").toBuilder().revision(1).build();
 
         IllegalArgumentException illegalArgumentException = Assertions.assertThrows(IllegalArgumentException.class, () -> graphService.flowGraph(flow, Collections.singletonList("root.launch")));
         assertThat(illegalArgumentException.getMessage()).isEqualTo("Can't expand subflow task 'launch' because namespace and/or flowId contains dynamic values. This can only be viewed on an execution.");
 
-        Execution execution = runnerUtils.runOne(MAIN_TENANT, "io.kestra.tests", "task-flow-dynamic", 1, (f, e) -> Map.of(
+        Execution execution = runnerUtils.runOne("tenant2", "io.kestra.tests", "task-flow-dynamic", 1, (f, e) -> Map.of(
             "namespace", f.getNamespace(),
             "flowId", "switch"
         ));
@@ -373,13 +372,17 @@ class FlowGraphTest {
     }
 
     private FlowWithSource parse(String path) throws IOException {
+        return parse(path, MAIN_TENANT);
+    }
+
+    private FlowWithSource parse(String path, String tenantId) throws IOException {
         URL resource = TestsUtils.class.getClassLoader().getResource(path);
         assert resource != null;
 
         File file = new File(resource.getFile());
 
         return YamlParser.parse(file, FlowWithSource.class).toBuilder()
-            .tenantId(MAIN_TENANT)
+            .tenantId(tenantId)
             .source(Files.readString(file.toPath()))
             .build();
     }

core/src/test/java/io/kestra/core/models/property/URIFetcherTest.java

@@ -4,6 +4,7 @@ import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.runners.*;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.IdUtils;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -84,8 +85,9 @@ class URIFetcherTest {
 
     @Test
     void shouldFetchFromNsfile() throws IOException {
-        URI uri = createNsFile(false);
-        RunContext runContext = runContextFactory.of(Map.of("flow", Map.of("namespace", "namespace")));
+        String namespace = IdUtils.create();
+        URI uri = createNsFile(namespace, false);
+        RunContext runContext = runContextFactory.of(Map.of("flow", Map.of("namespace", namespace)));
 
         try (var fetch = URIFetcher.of(uri).fetch(runContext)) {
             String fetchedContent = new String(fetch.readAllBytes());
@@ -95,7 +97,8 @@ class URIFetcherTest {
 
     @Test
     void shouldFetchFromNsfileFromOtherNs() throws IOException {
-        URI uri = createNsFile(true);
+        String namespace = IdUtils.create();
+        URI uri = createNsFile(namespace, true);
         RunContext runContext = runContextFactory.of(Map.of("flow", Map.of("namespace", "other")));
 
         try (var fetch = URIFetcher.of(uri).fetch(runContext)) {
@@ -139,8 +142,7 @@ class URIFetcherTest {
         );
     }
 
-    private URI createNsFile(boolean nsInAuthority) throws IOException {
-        String namespace = "namespace";
+    private URI createNsFile(String namespace, boolean nsInAuthority) throws IOException {
         String filePath = "file.txt";
         storage.createDirectory(MAIN_TENANT, namespace, URI.create(StorageContext.namespaceFilePrefix(namespace)));
         storage.put(MAIN_TENANT, namespace, URI.create(StorageContext.namespaceFilePrefix(namespace) + "/" + filePath), new ByteArrayInputStream("Hello World".getBytes()));

core/src/test/java/io/kestra/core/models/tasks/runners/ScriptServiceTest.java

@@ -10,6 +10,7 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.utils.IdUtils;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 
@@ -33,17 +34,15 @@ class ScriptServiceTest {
 
     @Test
     void replaceInternalStorage() throws IOException {
-        var runContext = runContextFactory.of();
+        String tenant = IdUtils.create();
+        var runContext = runContextFactory.of("id", "namespace", tenant);
         var command  = ScriptService.replaceInternalStorage(runContext, null, false);
         assertThat(command).isEqualTo("");
 
         command = ScriptService.replaceInternalStorage(runContext, "my command", false);
         assertThat(command).isEqualTo("my command");
 
-        Path path = Path.of("/tmp/unittest/main/file.txt");
-        if (!path.toFile().exists()) {
-            Files.createFile(path);
-        }
+        Path path = createFile(tenant, "file");
 
         String internalStorageUri = "kestra://some/file.txt";
         File localFile = null;
@@ -70,12 +69,10 @@ class ScriptServiceTest {
 
     @Test
     void replaceInternalStorageUnicode() throws IOException {
-        var runContext = runContextFactory.of();
+        String tenant = IdUtils.create();
+        var runContext = runContextFactory.of("id", "namespace", tenant);
 
-        Path path = Path.of("/tmp/unittest/main/file-龍.txt");
-        if (!path.toFile().exists()) {
-            Files.createFile(path);
-        }
+        Path path = createFile(tenant, "file-龍");
 
         String internalStorageUri = "kestra://some/file-龍.txt";
         File localFile = null;
@@ -95,12 +92,10 @@ class ScriptServiceTest {
 
     @Test
     void uploadInputFiles() throws IOException {
-        var runContext = runContextFactory.of();
+        String tenant = IdUtils.create();
+        var runContext = runContextFactory.of("id", "namespace", tenant);
 
-        Path path = Path.of("/tmp/unittest/main/file.txt");
-        if (!path.toFile().exists()) {
-            Files.createFile(path);
-        }
+        Path path = createFile(tenant, "file");
 
         List<File> filesToDelete = new ArrayList<>();
         String internalStorageUri = "kestra://some/file.txt";
@@ -143,13 +138,11 @@ class ScriptServiceTest {
 
     @Test
     void uploadOutputFiles() throws IOException {
-        var runContext = runContextFactory.of();
-        Path path = Path.of("/tmp/unittest/main/file.txt");
-        if (!path.toFile().exists()) {
-            Files.createFile(path);
-        }
+        String tenant = IdUtils.create();
+        var runContext = runContextFactory.of("id", "namespace", tenant);
+        Path path = createFile(tenant, "file");
 
-        var outputFiles = ScriptService.uploadOutputFiles(runContext, Path.of("/tmp/unittest/main"));
+        var outputFiles = ScriptService.uploadOutputFiles(runContext, Path.of("/tmp/unittest/%s".formatted(tenant)));
         assertThat(outputFiles, not(anEmptyMap()));
         assertThat(outputFiles.get("file.txt")).isEqualTo(URI.create("kestra:///file.txt"));
 
@@ -232,4 +225,13 @@ class ScriptServiceTest {
             .build();
         return runContextFactory.of(flow, task, execution, taskRun);
     }
+
+    private static Path createFile(String tenant, String fileName) throws IOException {
+        Path path = Path.of("/tmp/unittest/%s/%s.txt".formatted(tenant, fileName));
+        if (!path.toFile().exists()) {
+            Files.createDirectory(Path.of("/tmp/unittest/%s".formatted(tenant)));
+            Files.createFile(path);
+        }
+        return path;
+    }
 }

core/src/test/java/io/kestra/core/models/triggers/multipleflows/AbstractMultipleConditionStorageTest.java

@@ -3,6 +3,7 @@ package io.kestra.core.models.triggers.multipleflows;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.models.property.Property;
+import io.kestra.core.utils.TestsUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.junit.jupiter.api.Test;
 import io.kestra.plugin.core.condition.ExecutionFlow;
@@ -33,8 +34,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void allDefault() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
 
@@ -50,8 +52,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void daily() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().window(Duration.ofDays(1)).windowAdvance(Duration.ofSeconds(0)).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().window(Duration.ofDays(1)).windowAdvance(Duration.ofSeconds(0)).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
 
@@ -67,8 +70,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void dailyAdvance() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().window(Duration.ofDays(1)).windowAdvance(Duration.ofHours(4).negated()).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().window(Duration.ofDays(1)).windowAdvance(Duration.ofHours(4).negated()).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
 
@@ -84,8 +88,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void hourly() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().window(Duration.ofHours(1)).windowAdvance(Duration.ofHours(4).negated()).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().window(Duration.ofHours(1)).windowAdvance(Duration.ofHours(4).negated()).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
 
@@ -102,8 +107,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void minutely() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().window(Duration.ofMinutes(15)).windowAdvance(Duration.ofMinutes(5).negated()).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().window(Duration.ofMinutes(15)).windowAdvance(Duration.ofMinutes(5).negated()).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
 
@@ -115,8 +121,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void expiration() throws Exception {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().window(Duration.ofSeconds(2)).windowAdvance(Duration.ofMinutes(0).negated()).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().window(Duration.ofSeconds(2)).windowAdvance(Duration.ofMinutes(0).negated()).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
         this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
@@ -136,8 +143,9 @@ public abstract class AbstractMultipleConditionStorageTest {
     @Test
     void expired() throws Exception {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().window(Duration.ofSeconds(2)).windowAdvance(Duration.ofMinutes(0).negated()).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().window(Duration.ofSeconds(2)).windowAdvance(Duration.ofMinutes(0).negated()).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
         this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
@@ -146,20 +154,21 @@ public abstract class AbstractMultipleConditionStorageTest {
 
         assertThat(window.getResults().get("a")).isTrue();
 
-        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(null);
+        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isZero();
 
         Thread.sleep(2005);
 
-        expired = multipleConditionStorage.expired(null);
+        expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isEqualTo(1);
     }
 
     @Test
     void dailyTimeDeadline() throws Exception {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().type(Type.DAILY_TIME_DEADLINE).deadline(LocalTime.now().plusSeconds(2)).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().type(Type.DAILY_TIME_DEADLINE).deadline(LocalTime.now().plusSeconds(2)).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
         this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
@@ -168,20 +177,21 @@ public abstract class AbstractMultipleConditionStorageTest {
 
         assertThat(window.getResults().get("a")).isTrue();
 
-        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(null);
+        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isZero();
 
         Thread.sleep(2005);
 
-        expired = multipleConditionStorage.expired(null);
+        expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isEqualTo(1);
     }
 
     @Test
     void dailyTimeDeadline_Expired() throws Exception {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().type(Type.DAILY_TIME_DEADLINE).deadline(LocalTime.now().minusSeconds(1)).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().type(Type.DAILY_TIME_DEADLINE).deadline(LocalTime.now().minusSeconds(1)).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
         this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
@@ -190,16 +200,17 @@ public abstract class AbstractMultipleConditionStorageTest {
 
         assertThat(window.getResults()).isEmpty();
 
-        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(null);
+        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isEqualTo(1);
     }
 
     @Test
-    void dailyTimeWindow() throws Exception {
+    void dailyTimeWindow() {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
         LocalTime startTime = LocalTime.now().truncatedTo(ChronoUnit.MINUTES);
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().type(Type.DAILY_TIME_WINDOW).startTime(startTime).endTime(startTime.plusMinutes(5)).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().type(Type.DAILY_TIME_WINDOW).startTime(startTime).endTime(startTime.plusMinutes(5)).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
         this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
@@ -208,15 +219,16 @@ public abstract class AbstractMultipleConditionStorageTest {
 
         assertThat(window.getResults().get("a")).isTrue();
 
-        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(null);
+        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isZero();
     }
 
     @Test
     void slidingWindow() throws Exception {
         MultipleConditionStorageInterface multipleConditionStorage = multipleConditionStorage();
+        String tenant = TestsUtils.randomTenant(this.getClass().getSimpleName());
 
-        Pair<Flow, MultipleCondition> pair = mockFlow(TimeWindow.builder().type(Type.SLIDING_WINDOW).window(Duration.ofHours(1)).build());
+        Pair<Flow, MultipleCondition> pair = mockFlow(tenant, TimeWindow.builder().type(Type.SLIDING_WINDOW).window(Duration.ofHours(1)).build());
 
         MultipleConditionWindow window = multipleConditionStorage.getOrCreate(pair.getKey(), pair.getRight(), Collections.emptyMap());
         this.save(multipleConditionStorage, pair.getLeft(), Collections.singletonList(window.with(ImmutableMap.of("a", true))));
@@ -225,13 +237,13 @@ public abstract class AbstractMultipleConditionStorageTest {
 
         assertThat(window.getResults().get("a")).isTrue();
 
-        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(null);
+        List<MultipleConditionWindow> expired = multipleConditionStorage.expired(tenant);
         assertThat(expired.size()).isZero();
     }
 
-    private static Pair<Flow, MultipleCondition> mockFlow(TimeWindow sla) {
+    private static Pair<Flow, MultipleCondition> mockFlow(String tenantId, TimeWindow sla) {
         var multipleCondition = MultipleCondition.builder()
-            .id("condition-multiple")
+            .id("condition-multiple-%s".formatted(tenantId))
             .conditions(ImmutableMap.of(
                 "flow-a", ExecutionFlow.builder()
                     .flowId(Property.ofValue("flow-a"))
@@ -248,6 +260,7 @@ public abstract class AbstractMultipleConditionStorageTest {
         Flow flow = Flow.builder()
             .namespace(NAMESPACE)
             .id("multiple-flow")
+            .tenantId(tenantId)
             .revision(1)
             .triggers(Collections.singletonList(io.kestra.plugin.core.trigger.Flow.builder()
                 .id("trigger-flow")

core/src/test/java/io/kestra/core/reporter/reports/AbstractFeatureUsageReportTest.java

@@ -13,21 +13,20 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 @KestraTest
 public abstract class AbstractFeatureUsageReportTest {
-    
+
     @Inject
     FeatureUsageReport featureUsageReport;
-    
+
     @Test
     public void shouldGetReport() {
         // When
         Instant now = Instant.now();
         FeatureUsageReport.UsageEvent event = featureUsageReport.report(
-            now, 
+            now,
             Reportable.TimeInterval.of(now.minus(Duration.ofDays(1)).atZone(ZoneId.systemDefault()), now.atZone(ZoneId.systemDefault()))
         );
-        
+
         // Then
         assertThat(event.getExecutions().getDailyExecutionsCount().size()).isGreaterThan(0);
-        assertThat(event.getExecutions().getDailyTaskRunsCount()).isNull();
     }
 }

core/src/test/java/io/kestra/core/reporter/reports/AbstractServiceUsageReportTest.java

@@ -22,24 +22,24 @@ import java.util.Set;
 
 @KestraTest
 public abstract class AbstractServiceUsageReportTest {
-    
+
     @Inject
     ServiceUsageReport serviceUsageReport;
-    
+
     @Inject
     ServiceInstanceRepositoryInterface serviceInstanceRepository;
-    
+
     @Test
     public void shouldGetReport() {
         // Given
-        final LocalDate start = LocalDate.now().withDayOfMonth(1);
+        final LocalDate start = LocalDate.of(2025, 1, 1);
         final LocalDate end = start.withDayOfMonth(start.getMonth().length(start.isLeapYear()));
         final ZoneId zoneId = ZoneId.systemDefault();
-        
+
         LocalDate from = start;
         int days = 0;
         // generate one month of service instance
-        
+
         while (from.toEpochDay() < end.toEpochDay()) {
             Instant createAt = from.atStartOfDay(zoneId).toInstant();
             Instant updatedAt = from.atStartOfDay(zoneId).plus(Duration.ofHours(10)).toInstant();
@@ -62,14 +62,14 @@ public abstract class AbstractServiceUsageReportTest {
             from = from.plusDays(1);
             days++;
         }
-        
-        
+
+
         // When
         Instant now = end.plusDays(1).atStartOfDay(zoneId).toInstant();
         ServiceUsageReport.ServiceUsageEvent event = serviceUsageReport.report(now,
             Reportable.TimeInterval.of(start.atStartOfDay(zoneId), end.plusDays(1).atStartOfDay(zoneId))
         );
-        
+
         // Then
         List<ServiceUsage.DailyServiceStatistics> statistics = event.services().dailyStatistics();
         Assertions.assertEquals(ServiceType.values().length - 1, statistics.size());