try out to run in parallel JdbcRunnerRetryTest, lower exec run duration

* ci: make generated test report retrocompatible with older realeases

* ci: fix cli

.devcontainer/README.md

@@ -23,9 +23,19 @@ In the meantime, you can move onto the next step...
 
 ---
 
+### Requirements
+
+- Java 21 (LTS versions).
+  > ⚠️ Java 24 and above are not supported yet and will fail with `invalid source release: 21`.
+- Gradle (comes with wrapper `./gradlew`)
+- Docker (optional, for running Kestra in containers)
+
 ### Development:
 
-- Create a `.env.development.local` file in the `ui` folder and paste the following:
+- (Optional) By default, your dev server will target `localhost:8080`. If your backend is running elsewhere, you can create `.env.development.local` under `ui` folder with this content:
+```
+VITE_APP_API_URL={myApiUrl}
+```
 
 - Navigate into the `ui` folder and run `npm install` to install the dependencies for the frontend project.
 

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

@@ -2,7 +2,7 @@ name: Auto-Translate UI keys and create PR
 
 on:
   schedule:
-    - cron: "0 9-21 * * *"  # Every hour from 9 AM to 9 PM
+    - cron: "0 9-21/3 * * *"  # Every 3 hours from 9 AM to 9 PM
   workflow_dispatch:
     inputs:
       retranslate_modified_keys:
@@ -20,13 +20,13 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 10
     steps:
-        - uses: actions/checkout@v4
+        - uses: actions/checkout@v5
           name: Checkout
           with:
             fetch-depth: 0
 
         - name: Set up Python
-          uses: actions/setup-python@v5
+          uses: actions/setup-python@v6
           with:
             python-version: "3.x"
 
@@ -39,7 +39,7 @@ jobs:
             OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
 
         - name: Set up Node
-          uses: actions/setup-node@v4
+          uses: actions/setup-node@v5
           with:
             node-version: "20.x"
 

.github/workflows/codeql-analysis.yml

@@ -27,7 +27,7 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v4
+      uses: actions/checkout@v5
       with:
         # We must fetch at least the immediate parents so that if this is
         # a pull request then we can checkout the head.
@@ -50,7 +50,7 @@ jobs:
 
     # Set up JDK
     - name: Set up JDK
-      uses: actions/setup-java@v4
+      uses: actions/setup-java@v5
       if: ${{ matrix.language == 'java' }}
       with:
         distribution: 'temurin'

.github/workflows/docker.yml

@@ -1,178 +0,0 @@
-name: Create Docker images on Release
-
-on:
-  workflow_dispatch:
-    inputs:
-      retag-latest:
-        description: 'Retag latest Docker images'
-        required: true
-        type: string
-        default: "false"
-        options:
-          - "true"
-          - "false"
-      release-tag:
-        description: 'Kestra Release Tag'
-        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: string
-        default: "true"
-        options:
-          - "true"
-          - "false"
-
-env:
-  PLUGIN_VERSION: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
-jobs:
-  plugins:
-    name: List Plugins
-    runs-on: ubuntu-latest
-    outputs:
-      plugins: ${{ steps.plugins.outputs.plugins }}
-    steps:
-      # Checkout
-      - uses: actions/checkout@v4
-
-      # Get Plugins List
-      - name: Get Plugins List
-        uses: ./.github/actions/plugins-list
-        id: plugins
-        with:
-          plugin-version: ${{ env.PLUGIN_VERSION }}
-
-  # ********************************************************************************************************************
-  # Build
-  # ********************************************************************************************************************
-  build-artifacts:
-    name: Build Artifacts
-    if: ${{ github.event.inputs.force-download-artifact == 'true' }}
-    uses: ./.github/workflows/workflow-build-artifacts.yml
-
-  docker:
-    name: Publish Docker
-    needs: [ plugins, build-artifacts ]
-    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@v4
-
-      # 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 [[ "${{ 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
-
-      # [workflow_dispatch]
-      # Download executable from GitHub Release
-      - name: Artifacts - Download release (workflow_dispatch)
-        id: download-github-release
-        if: github.event_name == 'workflow_dispatch' && github.event.inputs.force-download-artifact == 'false'
-        uses: robinraju/release-downloader@v1.12
-        with:
-          tag: ${{steps.vars.outputs.tag}}
-          fileName: 'kestra-*'
-          out-file-path: build/executable
-
-      # [workflow_call]
-      # Download executable from artifact
-      - name: Artifacts - Download executable
-        if: github.event_name != 'workflow_dispatch' || steps.download-github-release.outcome == 'skipped'
-        uses: actions/download-artifact@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: github.event.inputs.retag-latest == 'true'
-        uses: regclient/actions/regctl-installer@main
-
-      - name: Retag to latest
-        if: github.event.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) }}
-
-  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/e2e.yml

@@ -19,7 +19,7 @@ on:
         default: "no input"
 jobs:
   check:
-    timeout-minutes: 10
+    timeout-minutes: 15
     runs-on: ubuntu-latest
     env:
       GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
@@ -32,12 +32,12 @@ jobs:
           password: ${{ github.token }}
 
       - name: Checkout kestra
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           path: kestra
 
       # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
+      - uses: kestra-io/actions/composite/setup-build@main
         name: Setup - Build
         id: build
         with:

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

@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
     inputs:
       releaseVersion:
-        description: 'The release version (e.g., 0.21.0-rc1)'
+        description: 'The release version (e.g., 0.21.0)'
         required: true
         type: string
       nextVersion:
@@ -21,25 +21,17 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       # Checkout
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v4
-        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
@@ -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,10 +65,10 @@ 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}} \
           --dry-run \
           --yes \
-          ${{ steps.plugins-list.outputs.repositories }}
+          ${{ steps.plugins-list.outputs.repositories }}

.github/workflows/gradle-release.yml

@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
     inputs:
       releaseVersion:
-        description: 'The release version (e.g., 0.21.0-rc1)'
+        description: 'The release version (e.g., 0.21.0)'
         required: true
         type: string
       nextVersion:
@@ -23,8 +23,8 @@ jobs:
       # Checks
       - name: Check Inputs
         run: |
-          if ! [[ "$RELEASE_VERSION" =~ ^[0-9]+(\.[0-9]+)\.0-rc[01](-SNAPSHOT)?$ ]]; then
-            echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)\.0-rc[01](-SNAPSHOT)?$"
+          if ! [[ "$RELEASE_VERSION" =~ ^[0-9]+(\.[0-9]+)\.0$ ]]; then
+            echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)\.0$"
             exit 1
           fi
 
@@ -33,20 +33,13 @@ jobs:
             exit 1;
           fi
       # Checkout
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           fetch-depth: 0
           path: kestra
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v4
-        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
@@ -78,7 +71,6 @@ jobs:
           git checkout develop;
 
           if [[ "$RELEASE_VERSION" == *"-SNAPSHOT" ]]; then
-            # -SNAPSHOT qualifier maybe used to test release-candidates
             ./gradlew release -Prelease.useAutomaticVersion=true \
             -Prelease.releaseVersion="${RELEASE_VERSION}" \
             -Prelease.newVersion="${NEXT_VERSION}" \
@@ -89,4 +81,4 @@ jobs:
             -Prelease.releaseVersion="${RELEASE_VERSION}" \
             -Prelease.newVersion="${NEXT_VERSION}" \
             -Prelease.pushReleaseVersionBranch="${PUSH_RELEASE_BRANCH}"
-          fi
+          fi

.github/workflows/kestra-devtools-test.yml

@@ -0,0 +1,32 @@
+name: kestra-devtools test
+
+on:
+  pull_request:
+    branches:
+      - develop
+    paths:
+      - 'dev-tools/kestra-devtools/**'
+
+env:
+  # to save corepack from itself
+  COREPACK_INTEGRITY_KEYS: 0
+
+jobs:
+  test:
+    name: kestra-devtools tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v5
+
+      - name: Npm - install
+        working-directory: 'dev-tools/kestra-devtools'
+        run: npm ci
+
+      - name: Run tests
+        working-directory: 'dev-tools/kestra-devtools'
+        run: npm run test
+
+      - name: Npm - Run build
+        working-directory: 'dev-tools/kestra-devtools'
+        run: npm run build

.github/workflows/main.yml

@@ -3,10 +3,17 @@ name: Main Workflow
 on:
   workflow_dispatch:
     inputs:
+      skip-test:
+        description: 'Skip test'
+        type: choice
+        required: true
+        default: 'false'
+        options:
+          - "true"
+          - "false"
       plugin-version:
-          description: "Kestra version"
-          default: 'LATEST'
-          required: true
+          description: "plugins version"
+          required: false
           type: string
   push:
     branches:
@@ -25,16 +32,17 @@ 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: "!startsWith(github.ref, 'refs/heads/releases')"
+    if: "!failure() && !cancelled() && !startsWith(github.ref, 'refs/heads/releases')"
     uses: ./.github/workflows/workflow-release.yml
     with:
-        plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
+        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 }}
@@ -45,13 +53,12 @@ jobs:
       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
@@ -61,14 +68,9 @@ jobs:
           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 }}
+        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:
-          repo_token: ${{ secrets.GITHUB_TOKEN }}
-          slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
-          name: GitHub Actions
-          icon_emoji: ":github-actions:"
-          channel: "C02DQ1A7JLR" # _int_git channel
+          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}

.github/workflows/pull-request.yml

@@ -60,19 +60,3 @@ jobs:
     name: E2E - Tests
     uses: ./.github/workflows/e2e.yml
 
-  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"

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

@@ -17,7 +17,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       # Checkout
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           fetch-depth: 0
 

.github/workflows/setversion-tag.yml

@@ -34,11 +34,14 @@ jobs:
           fi
 
       # Checkout
-      - uses: actions/checkout@v4
+      - name: Checkout
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
+          token: ${{ secrets.GH_PERSONAL_TOKEN }}
 
-      - name: Configure Git
+      # Configure
+      - name: Git - Configure
         run: |
           git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com"
           git config --global user.name "github-actions[bot]"
@@ -54,4 +57,4 @@ jobs:
           git commit -m"chore(version): update to version '$RELEASE_VERSION'"
           git push
           git tag -a "v$RELEASE_VERSION" -m"v$RELEASE_VERSION"
-          git push --tags
+          git push --tags

.github/workflows/vulnerabilities-check.yml

@@ -17,17 +17,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       # Checkout
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v4
-        with:
-          repository: kestra-io/actions
-          path: actions
-          ref: main
-
       # Setup build
       - uses: ./actions/.github/actions/setup-build
         id: build
@@ -66,19 +59,12 @@ jobs:
       actions: read
     steps:
       # Checkout
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v4
-        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
@@ -87,7 +73,7 @@ jobs:
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.32.0
+        uses: aquasecurity/trivy-action@0.33.1
         with:
           image-ref: kestra/kestra:develop
           format: 'template'
@@ -111,28 +97,20 @@ jobs:
       actions: read
     steps:
       # Checkout
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         with:
           fetch-depth: 0
 
-      # Checkout GitHub Actions
-      - uses: actions/checkout@v4
-        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
-        uses: aquasecurity/trivy-action@0.32.0
+        uses: aquasecurity/trivy-action@0.33.1
         with:
           image-ref: kestra/kestra:latest
           format: table

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

@@ -20,6 +20,7 @@ permissions:
   contents: write
   checks: write
   actions: read
+  pull-requests: write
 
 jobs:
   test:
@@ -29,13 +30,13 @@ jobs:
       GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
       SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         name: Checkout - Current ref
         with:
           fetch-depth: 0
 
       # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
+      - uses: kestra-io/actions/composite/setup-build@main
         name: Setup - Build
         id: build
         with:
@@ -59,6 +60,30 @@ jobs:
           export GOOGLE_APPLICATION_CREDENTIALS=$HOME/.gcp-service-account.json
           ./gradlew check javadoc --parallel
 
+      - name: comment PR with test report
+        if: ${{ !cancelled() && github.event_name == 'pull_request' }}
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}
+        run: |
+          export KESTRA_PWD=$(pwd) && sh -c 'cd dev-tools/kestra-devtools && npm ci && npm run build && node dist/kestra-devtools-cli.cjs generateTestReportSummary --only-errors --ci $KESTRA_PWD' > report.md
+          cat report.md
+      # Gradle check
+      - name: 'generate Timeline flamegraph'
+        if: always()
+        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 mergeTestTimeline
+      - name: 'Upload Timeline flamegraph'
+        uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: all-test-timelines.json
+          path: build/reports/test-timelines-report/all-test-timelines.json
+          retention-days: 5
       # report test
       - name: Test - Publish Test Results
         uses: dorny/test-reporter@v2
@@ -86,13 +111,13 @@ jobs:
         id: auth
         if: always() && env.GOOGLE_SERVICE_ACCOUNT != ''
         continue-on-error: true
-        uses: "google-github-actions/auth@v2"
+        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@v2"
+        uses: "google-github-actions/setup-gcloud@v3"
 
         # Allure check
       - uses: rlespinasse/github-slug-action@v5

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

@@ -15,7 +15,7 @@ jobs:
       PLUGIN_VERSION: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
     steps:
       - name: Checkout - Current ref
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
 
@@ -26,7 +26,7 @@ jobs:
         run: npm ci
 
       # Setup build
-      - uses: kestra-io/actions/.github/actions/setup-build@main
+      - uses: kestra-io/actions/composite/setup-build@main
         name: Setup - Build
         id: build
         with:

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

@@ -20,7 +20,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Cache Node Modules
         id: cache-node-modules

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

@@ -20,20 +20,11 @@ jobs:
     steps:
       # Check out
       - name: Checkout - Repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
           submodules: true
 
-      # Checkout GitHub Actions
-      - name: Checkout - Actions
-        uses: actions/checkout@v4
-        with:
-          repository: kestra-io/actions
-          sparse-checkout-cone-mode: true
-          path: actions
-          sparse-checkout: |
-            .github/actions
 
       # Download Exec
       # Must be done after checkout actions
@@ -59,7 +50,7 @@ jobs:
 
       # GitHub Release
       - name: Create GitHub release
-        uses: ./actions/.github/actions/github-release
+        uses: kestra-io/actions/composite/github-release@main
         if: ${{ startsWith(github.ref, 'refs/tags/v') }}
         env:
           MAKE_LATEST: ${{ steps.is_latest.outputs.latest }}
@@ -78,4 +69,11 @@ jobs:
               "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: kestra-io/actions/composite/github-release-note-merge@main
+        env:
+            GITHUB_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+            RELEASE_TAG: ${{ github.ref_name }}

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

@@ -1,22 +1,45 @@
-name: Publish - Docker
+name: Create Docker images on Release
 
 on:
   workflow_dispatch:
     inputs:
-      plugin-version:
-        description: "Kestra version"
-        default: 'LATEST'
+      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: string
+        type: choice
         default: "true"
+        options:
+          - "true"
+          - "false"
   workflow_call:
     inputs:
       plugin-version:
-        description: "Kestra version"
+        description: "Plugin version"
         default: 'LATEST'
         required: false
         type: string
@@ -33,45 +56,93 @@ on:
         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: ${{ github.event.inputs.force-download-artifact == 'true' }}
+    if: ${{ inputs.force-download-artifact == 'true' }}
     uses: ./.github/workflows/workflow-build-artifacts.yml
-  # ********************************************************************************************************************
-  # Docker
-  # ********************************************************************************************************************
-  publish:
-    name: Publish - Docker
+
+  docker:
+    name: Publish Docker
+    needs: [ plugins, build-artifacts ]
+    if: always()
     runs-on: ubuntu-latest
-    needs: build-artifacts
-    if: |
-      always() &&
-      (needs.build-artifacts.result == 'success' ||
-        github.event.inputs.force-download-artifact != 'true')
-    env:
-      PLUGIN_VERSION: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
     strategy:
       matrix:
         image:
-          - tag: -no-plugins
+          - name: "-no-plugins"
+            plugins: ""
             packages: jattach
-            plugins: false
-            python-libraries: ""
-
-          - tag: ""
-            plugins: true
-            packages: python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach
-            python-libraries: kestra
+            python-libs: ""
+          - name: ""
+            plugins: ${{needs.plugins.outputs.plugins}}
+            packages: python3 python-is-python3 python3-pip curl jattach
+            python-libs: kestra
     steps:
-      - name: Checkout - Current ref
-        uses: actions/checkout@v4
+      - 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: Docker - Setup QEMU
+      - name: Set up QEMU
         uses: docker/setup-qemu-action@v3
 
       - name: Docker - Fix Qemu
@@ -79,66 +150,59 @@ jobs:
         run: |
           docker run --rm --privileged multiarch/qemu-user-static --reset -p yes -c yes
 
-      - name: Docker - Setup Docker Buildx
+      - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 
       # Docker Login
-      - name: Docker - Login to DockerHub
+      - name: Login to DockerHub
         uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_PASSWORD }}
 
-
-#     # Get Plugins List
-      - name: Plugins - Get List
-        uses: ./.github/actions/plugins-list
-        id: plugins-list
-        if: ${{ matrix.image.plugins}}
-        with:
-          plugin-version: ${{ env.PLUGIN_VERSION }}
-
-      # Vars
-      - name: Docker - Set variables
-        shell: bash
-        id: vars
-        run: |
-          TAG=${GITHUB_REF#refs/*/}
-          PLUGINS="${{ matrix.image.plugins == true && steps.plugins-list.outputs.plugins || '' }}"
-          if [[ $TAG == v* ]]; then
-            TAG="${TAG}";
-            echo "plugins=${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT
-          elif [[ $TAG = "develop" ]]; then
-            TAG="develop";
-            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
-          else
-            TAG="build-${{ github.run_id }}";
-            echo "plugins=--repositories=https://central.sonatype.com/repository/maven-snapshots/ $PLUGINS" >> $GITHUB_OUTPUT
-          fi
-
-          echo "tag=${TAG}${{ matrix.image.tag }}" >> $GITHUB_OUTPUT
-
-      # Build Docker Image
-      - 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
-
       # Docker Build and push
-      - name: Docker - Build image
+      - name: Push to Docker Hub
         uses: docker/build-push-action@v6
         with:
           context: .
           push: true
-          tags: kestra/kestra:${{ steps.vars.outputs.tag }}
+          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-libraries }}
+            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:
+      - name: Slack notification
+        if: ${{ failure() && env.SLACK_WEBHOOK_URL != 0 }}
+        uses: kestra-io/actions/composite/slack-status@main
+        with:
+          webhook-url: ${{ secrets.SLACK_WEBHOOK_URL }}
+

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

@@ -25,11 +25,11 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout - Current ref
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       # Setup build
       - name: Setup - Build
-        uses: kestra-io/actions/.github/actions/setup-build@main
+        uses: kestra-io/actions/composite/setup-build@main
         id: build
         with:
           java-enabled: true

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

@@ -7,6 +7,7 @@ on:
 jobs:
   publish:
     name: Pull Request - Delete Docker
+    if: github.repository == 'kestra-io/kestra' # prevent running on forks
     runs-on: ubuntu-latest
     steps:
       - uses: dataaxiom/ghcr-cleanup-action@v1

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

@@ -8,17 +8,19 @@ on:
 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@v4
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
 
@@ -60,7 +62,7 @@ jobs:
 
       # Add comment on pull request
       - name: Add comment to PR
-        uses: actions/github-script@v7
+        uses: actions/github-script@v8
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |

.github/workflows/workflow-release.yml

@@ -4,7 +4,7 @@ on:
   workflow_dispatch:
     inputs:
       plugin-version:
-        description: "Kestra version"
+        description: "plugins version"
         default: 'LATEST'
         required: false
         type: string
@@ -16,7 +16,7 @@ on:
   workflow_call:
     inputs:
       plugin-version:
-        description: "Kestra version"
+        description: "plugins version"
         default: 'LATEST'
         required: false
         type: string
@@ -57,10 +57,10 @@ jobs:
     name: Publish Docker
     needs: build-artifacts
     uses: ./.github/workflows/workflow-publish-docker.yml
-    if: startsWith(github.ref, 'refs/heads/develop') || github.event.inputs.publish-docker == 'true'
+    if: github.ref == 'refs/heads/develop' || inputs.publish-docker == 'true'
     with:
       force-download-artifact: 'false'
-      plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
+      plugin-version: ${{ inputs.plugin-version != null && inputs.plugin-version || 'LATEST' }}
     secrets:
       DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
       DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}

.github/workflows/workflow-test.yml

@@ -27,7 +27,7 @@ jobs:
       ui: ${{ steps.changes.outputs.ui }}
       backend: ${{ steps.changes.outputs.backend }}
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
         if: "!startsWith(github.ref, 'refs/tags/v')"
       - uses: dorny/paths-filter@v3
         if: "!startsWith(github.ref, 'refs/tags/v')"
@@ -84,14 +84,12 @@ jobs:
     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 }}
+          always() &&
+            (needs.frontend.result != 'success' || needs.backend.result != 'success') &&
+            (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 }}

.plugins

@@ -36,6 +36,7 @@
 #plugin-gemini:io.kestra.plugin:plugin-gemini:LATEST
 #plugin-git:io.kestra.plugin:plugin-git:LATEST
 #plugin-github:io.kestra.plugin:plugin-github:LATEST
+#plugin-gitlab:io.kestra.plugin:plugin-gitlab:LATEST
 #plugin-googleworkspace:io.kestra.plugin:plugin-googleworkspace:LATEST
 #plugin-graalvm:io.kestra.plugin:plugin-graalvm:LATEST
 #plugin-graphql:io.kestra.plugin:plugin-graphql:LATEST
@@ -108,16 +109,17 @@
 #plugin-serdes:io.kestra.plugin:plugin-serdes:LATEST
 #plugin-servicenow:io.kestra.plugin:plugin-servicenow:LATEST
 #plugin-sifflet:io.kestra.plugin:plugin-sifflet:LATEST
-#plugin-singer:io.kestra.plugin:plugin-singer:LATEST
 #plugin-soda:io.kestra.plugin:plugin-soda:LATEST
 #plugin-solace:io.kestra.plugin:plugin-solace:LATEST
 #plugin-spark:io.kestra.plugin:plugin-spark:LATEST
 #plugin-sqlmesh:io.kestra.plugin:plugin-sqlmesh:LATEST
+#plugin-supabase:io.kestra.plugin:plugin-supabase:LATEST
 #plugin-surrealdb:io.kestra.plugin:plugin-surrealdb:LATEST
 #plugin-terraform:io.kestra.plugin:plugin-terraform:LATEST
 #plugin-transform:io.kestra.plugin:plugin-transform-grok:LATEST
 #plugin-transform:io.kestra.plugin:plugin-transform-json:LATEST
 #plugin-tika:io.kestra.plugin:plugin-tika:LATEST
+#plugin-trivy:io.kestra.plugin:plugin-trivy:LATEST
 #plugin-weaviate:io.kestra.plugin:plugin-weaviate:LATEST
 #plugin-zendesk:io.kestra.plugin:plugin-zendesk:LATEST
 #plugin-typesense:io.kestra.plugin:plugin-typesense:LATEST

AGENTS.md

@@ -0,0 +1,305 @@
+# Kestra AGENTS.md
+
+This file provides guidance for AI coding agents working on the Kestra project. Kestra is an open-source data orchestration and scheduling platform built with Java (Micronaut) and Vue.js.
+
+## Repository Layout
+
+- **`core/`**: Core Kestra framework and task definitions
+- **`cli/`**: Command-line interface and server implementation
+- **`webserver/`**: REST API server implementation
+- **`ui/`**: Vue.js frontend application
+- **`jdbc-*`**: Database connector modules (H2, MySQL, PostgreSQL)
+- **`script/`**: Script execution engine
+- **`storage-local/`**: Local file storage implementation
+- **`repository-memory/`**: In-memory repository implementation
+- **`runner-memory/`**: In-memory execution runner
+- **`processor/`**: Task processing engine
+- **`model/`**: Data models and Data Transfer Objects
+- **`platform/`**: Platform-specific implementations
+- **`tests/`**: Integration test framework
+- **`e2e-tests/`**: End-to-end testing suite
+
+## Development Environment
+
+### Prerequisites
+
+- Java 21+
+- Node.js 22+ and npm
+- Python 3, pip, and python venv
+- Docker & Docker Compose
+- Gradle (wrapper included)
+
+### Quick Setup with Devcontainer
+
+The easiest way to get started is using the provided devcontainer:
+
+1. Install VSCode Remote Development extension
+2. Run `Dev Containers: Open Folder in Container...` from command palette
+3. Select the Kestra root folder
+4. Wait for Gradle build to complete
+
+### Manual Setup
+
+1. Clone the repository
+2. Run `./gradlew build` to build the backend
+3. Navigate to `ui/` and run `npm install`
+4. Create configuration files as described below
+
+## Configuration Files
+
+### Backend Configuration
+
+Create `cli/src/main/resources/application-override.yml`:
+
+**Local Mode (H2 database):**
+
+```yaml
+micronaut:
+  server:
+    cors:
+      enabled: true
+      configurations:
+        all:
+          allowedOrigins:
+            - http://localhost:5173
+```
+
+**Standalone Mode (PostgreSQL):**
+
+```yaml
+kestra:
+  repository:
+    type: postgres
+  storage:
+    type: local
+    local:
+      base-path: "/app/storage"
+  queue:
+    type: postgres
+  tasks:
+    tmp-dir:
+      path: /tmp/kestra-wd/tmp
+  anonymous-usage-report:
+    enabled: false
+
+datasources:
+  postgres:
+    url: jdbc:postgresql://host.docker.internal:5432/kestra
+    driverClassName: org.postgresql.Driver
+    username: kestra
+    password: k3str4
+
+flyway:
+  datasources:
+    postgres:
+      enabled: true
+      locations:
+        - classpath:migrations/postgres
+      ignore-migration-patterns: "*:missing,*:future"
+      out-of-order: true
+
+micronaut:
+  server:
+    cors:
+      enabled: true
+      configurations:
+        all:
+          allowedOrigins:
+            - http://localhost:5173
+```
+
+### Frontend Configuration
+
+Create `ui/.env.development.local` for environment variables.
+
+## Running the Application
+
+### Backend
+
+- **Local mode**: `./gradlew runLocal` (uses H2 database)
+- **Standalone mode**: Use VSCode Run and Debug with main class `io.kestra.cli.App` and args `server standalone`
+
+### Frontend
+
+- Navigate to `ui/` directory
+- Run `npm run dev` for development server (port 5173)
+- Run `npm run build` for production build
+
+## Building and Testing
+
+### Backend
+
+```bash
+# Build the project
+./gradlew build
+
+# Run tests
+./gradlew test
+
+# Run specific module tests
+./gradlew :core:test
+
+# Clean build
+./gradlew clean build
+```
+
+### Frontend
+
+```bash
+cd ui
+npm install
+npm run test
+npm run lint
+npm run build
+```
+
+### End-to-End Tests
+
+```bash
+# Build and start E2E tests
+./build-and-start-e2e-tests.sh
+
+# Or use the Makefile
+make install
+make install-plugins
+make start-standalone-postgres
+```
+
+## Development Guidelines
+
+### Java Backend
+
+- Use Java 21 features
+- Follow Micronaut framework patterns
+- Add Swagger annotations for API documentation
+- Use annotation processors (enable in IDE)
+- Set `MICRONAUT_ENVIRONMENTS=local,override` for custom config
+- Set `KESTRA_PLUGINS_PATH` for custom plugin loading
+
+### Vue.js Frontend
+
+- Vue 3 with Composition API
+- TypeScript for type safety
+- Vite for build tooling
+- ESLint and Prettier for code quality
+- Component-based architecture in `src/components/`
+
+### Code Style
+
+- Follow `.editorconfig` settings
+- Use 4 spaces for Java, 2 spaces for YAML/JSON/CSS
+- Enable format on save in VSCode
+- Use Prettier for frontend code formatting
+
+## Testing Strategy
+
+### Backend Testing
+
+- Unit tests in `src/test/java/`
+- Integration tests in `tests/` module
+- Use Micronaut test framework
+- Test both local and standalone modes
+
+### Frontend Testing
+- Unit tests with Jest
+- E2E tests with Playwright
+- Component testing with Storybook
+- Run `npm run test:unit` and `npm run test:e2e`
+
+## Plugin Development
+
+### Creating Plugins
+
+- Follow the [Plugin Developer Guide](https://kestra.io/docs/plugin-developer-guide/)
+- Place JAR files in `KESTRA_PLUGINS_PATH`
+- Use the plugin template structure
+- Test with both local and standalone modes
+
+### Plugin Loading
+
+- Set `KESTRA_PLUGINS_PATH` environment variable
+- Use devcontainer mounts for local development
+- Plugins are loaded at startup
+
+## Common Issues and Solutions
+
+### JavaScript Heap Out of Memory
+
+Set `NODE_OPTIONS=--max-old-space-size=4096` environment variable.
+
+### CORS Issues
+
+Ensure backend CORS is configured for `http://localhost:5173` when using frontend dev server.
+
+### Database Connection Issues
+
+- Use `host.docker.internal` instead of `localhost` when connecting from devcontainer
+- Verify PostgreSQL is running and accessible
+- Check database credentials and permissions
+
+### Gradle Build Issues
+
+- Clear Gradle cache: `./gradlew clean`
+- Check Java version compatibility
+- Verify all dependencies are available
+
+## Pull Request Guidelines
+
+### Before Submitting
+
+1. Run all tests: `./gradlew test` and `npm test`
+2. Check code formatting: `./gradlew spotlessCheck`
+3. Verify CORS configuration if changing API
+4. Test both local and standalone modes
+5. Update documentation for user-facing changes
+
+### Commit Messages
+
+- Follow conventional commit format
+- Use present tense ("Add feature" not "Added feature")
+- Reference issue numbers when applicable
+- Keep commits focused and atomic
+
+### Review Checklist
+
+- [ ] All tests pass
+- [ ] Code follows project style guidelines
+- [ ] Documentation is updated
+- [ ] No breaking changes without migration guide
+- [ ] CORS properly configured if API changes
+- [ ] Both local and standalone modes tested
+
+## Useful Commands
+
+```bash
+# Quick development commands
+./gradlew runLocal                    # Start local backend
+./gradlew :ui:build                   # Build frontend
+./gradlew clean build                 # Clean rebuild
+npm run dev                           # Start frontend dev server
+make install                          # Install Kestra locally
+make start-standalone-postgres        # Start with PostgreSQL
+
+# Testing commands
+./gradlew test                        # Run all backend tests
+./gradlew :core:test                  # Run specific module tests
+npm run test                          # Run frontend tests
+npm run lint                          # Lint frontend code
+```
+
+## Getting Help
+
+- Open a [GitHub issue](https://github.com/kestra-io/kestra/issues)
+- Join the [Kestra Slack community](https://kestra.io/slack)
+- Check the [main documentation](https://kestra.io/docs)
+
+## Environment Variables
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `MICRONAUT_ENVIRONMENTS` | Custom config environments | `local,override` |
+| `KESTRA_PLUGINS_PATH` | Path to custom plugins | `/workspaces/kestra/local/plugins` |
+| `NODE_OPTIONS` | Node.js options | `--max-old-space-size=4096` |
+| `JAVA_HOME` | Java installation path | `/usr/java/jdk-21` |
+
+Remember: Always test your changes in both local and standalone modes, and ensure CORS is properly configured for frontend development.

Makefile

@@ -89,7 +89,7 @@ build-docker: build-exec
 		--compress \
 		--rm \
 		-f ./Dockerfile \
-		--build-arg="APT_PACKAGES=python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach" \
+		--build-arg="APT_PACKAGES=python3 python-is-python3 python3-pip curl jattach" \
 		--build-arg="PYTHON_LIBRARIES=kestra" \
 		-t ${DOCKER_IMAGE}:${VERSION} ${DOCKER_PATH} || exit 1 ;
 

build-and-start-e2e-tests.sh

@@ -7,7 +7,7 @@ set -e
 # run tests on this image
 
 
-LOCAL_IMAGE_VERSION="local-e2e"
+LOCAL_IMAGE_VERSION="local-e2e-$(date +%s)"
 
 echo "Running E2E"
 echo "Start time: $(date '+%Y-%m-%d %H:%M:%S')"
@@ -15,6 +15,7 @@ start_time=$(date +%s)
 
 echo ""
 echo "Building the image for this current repository"
+make clean
 make build-docker VERSION=$LOCAL_IMAGE_VERSION
 
 end_time=$(date +%s)
@@ -32,7 +33,7 @@ echo "npm i"
 npm i
 
 echo 'sh ./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"'
-sh ./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"
+./run-e2e-tests.sh --kestra-docker-image-to-test "kestra/kestra:$LOCAL_IMAGE_VERSION"
 
 end_time2=$(date +%s)
 elapsed2=$(( end_time2 - start_time2 ))

build.gradle

@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "6.2.0.5505"
+    id "org.sonarqube" version "6.3.1.5724"
     id 'jacoco-report-aggregation'
 
     // helper
@@ -32,7 +32,7 @@ plugins {
 
     // release
     id 'net.researchgate.release' version '3.1.0'
-    id "com.gorylenko.gradle-git-properties" version "2.5.2"
+    id "com.gorylenko.gradle-git-properties" version "2.5.3"
     id 'signing'
     id "com.vanniktech.maven.publish" version "0.34.0"
 
@@ -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,52 @@ subprojects {
             environment 'SECRET_PASSWORD', "cGFzc3dvcmQ="
             environment 'ENV_TEST1', "true"
             environment 'ENV_TEST2', "Pass by env"
+
+            // === Test Timeline Trace (Chrome trace format) ===
+            // Produces per-JVM ndjson under build/test-timelines/*.jsonl and a merged array via :mergeTestTimeline
+            // Each event has: start time (ts, µs since epoch), end via dur, and absolute duration (dur, µs)
+            doFirst {
+                file("${buildDir}/test-results/test-timelines").mkdirs()
+            }
+
+            def jvmName = java.lang.management.ManagementFactory.runtimeMXBean.name
+            def pid = jvmName.tokenize('@')[0]
+            def traceDir = file("${buildDir}/test-results/test-timelines")
+            def traceFile = new File(traceDir, "${project.name}-${name}-${pid}.jsonl")
+            def starts = new java.util.concurrent.ConcurrentHashMap<Object, Long>()
+
+            beforeTest { org.gradle.api.tasks.testing.TestDescriptor d ->
+                // epoch millis to allow cross-JVM merge
+                starts.put(d, System.currentTimeMillis())
+            }
+            afterTest { org.gradle.api.tasks.testing.TestDescriptor d, org.gradle.api.tasks.testing.TestResult r ->
+                def st = starts.remove(d)
+                if (st != null) {
+                    def en = System.currentTimeMillis()
+                    long tsMicros  = st * 1000L          // start time (µs since epoch)
+                    long durMicros = (en - st) * 1000L   // duration (µs)
+                    def ev = [
+                        name: (d.className ? d.className + '.' + d.name : d.name),
+                        cat : 'test',
+                        ph  : 'X',              // Complete event with duration
+                        ts  : tsMicros,
+                        dur : durMicros,
+                        pid : project.name,     // group by project/module
+                        tid : "${name}-worker-${pid}",
+                        args: [result: r.resultType.toString()]
+                    ]
+                    synchronized (traceFile.absolutePath.intern()) {
+                        traceFile << (groovy.json.JsonOutput.toJson(ev) + System.lineSeparator())
+                    }
+                }
+            }
+            if (subProj.name == 'core' || subProj.name == 'jdbc-h2') {
+                // 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 {
@@ -236,7 +290,53 @@ subprojects {
         }
     }
 }
+// Root-level aggregator: merge timelines from ALL modules into one Chrome trace
+if (project == rootProject) {
+    tasks.register('mergeTestTimeline') {
+        group = 'verification'
+        description = 'Merge per-worker test timeline ndjson from all modules into a single Chrome Trace JSON array.'
+        doLast {
+            def collectedFiles = [] as List<File>
 
+            // Collect *.jsonl files from every subproject
+            rootProject.subprojects.each { p ->
+                def dir = p.file("${p.buildDir}/test-results/test-timelines")
+                if (dir.exists()) {
+                    collectedFiles.addAll(p.fileTree(dir: dir, include: '*.jsonl').files)
+                }
+            }
+
+            if (collectedFiles.isEmpty()) {
+                logger.lifecycle("No timeline files found in any subproject. Run tests first (e.g., './gradlew test --parallel').")
+                return
+            }
+
+            collectedFiles = collectedFiles.sort { it.name }
+
+            def outDir = rootProject.file("${rootProject.buildDir}/reports/test-timelines-report")
+            outDir.mkdirs()
+            def out = new File(outDir, "all-test-timelines.json")
+
+            out.withWriter('UTF-8') { w ->
+                w << '['
+                boolean first = true
+                collectedFiles.each { f ->
+                    f.eachLine { line ->
+                        def trimmed = line?.trim()
+                        if (trimmed) {
+                            if (!first) w << ','
+                            w << trimmed
+                            first = false
+                        }
+                    }
+                }
+                w << ']'
+            }
+
+            logger.lifecycle("Merged ${collectedFiles.size()} files into ${out} — open it in chrome://tracing or Perfetto UI.")
+        }
+    }
+}
 /**********************************************************************************************************************\
  * End-to-End Tests
  **********************************************************************************************************************/

cli/build.gradle

@@ -33,8 +33,13 @@ dependencies {
 
     implementation project(":storage-local")
 
+    // Kestra server components
+    implementation project(":executor")
+    implementation project(":scheduler")
     implementation project(":webserver")
+    implementation project(":worker")
 
     //test
+    testImplementation project(':tests')
     testImplementation "org.wiremock:wiremock-jetty12"
 }

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

@@ -89,11 +89,24 @@ public class App implements Callable<Integer> {
      */
     protected static ApplicationContext applicationContext(Class<?> mainClass,
                                                            String[] args) {
+        return applicationContext(mainClass, new String [] { Environment.CLI }, args);
+    }
+
+    /**
+     * Create an {@link ApplicationContext} with additional properties based on configuration files (--config) and
+     * forced Properties from current command.
+     *
+     * @param args args passed to java app
+     * @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/StandAloneRunner.java

@@ -1,15 +1,14 @@
-package io.kestra.core.runners;
+package io.kestra.cli;
 
-import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.core.runners.*;
 import io.kestra.core.server.Service;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
+import io.kestra.worker.DefaultWorker;
 import io.micronaut.context.ApplicationContext;
-import io.micronaut.context.annotation.Requires;
 import io.micronaut.context.annotation.Value;
 import jakarta.annotation.PreDestroy;
 import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -24,9 +23,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 @SuppressWarnings("try")
 @Slf4j
-@Singleton
-@Requires(missingBeans = RunnerInterface.class)
-public class StandAloneRunner implements RunnerInterface, AutoCloseable {
+public class StandAloneRunner implements Runnable, AutoCloseable {
     @Setter protected int workerThread = Math.max(3, Runtime.getRuntime().availableProcessors());
     @Setter protected boolean schedulerEnabled = true;
     @Setter protected boolean workerEnabled = true;
@@ -45,7 +42,7 @@ public class StandAloneRunner implements RunnerInterface, AutoCloseable {
 
     private final AtomicBoolean running = new AtomicBoolean(false);
 
-    private volatile ExecutorService poolExecutor;
+    private ExecutorService poolExecutor;
 
     @Override
     public void run() {
@@ -57,20 +54,20 @@ public class StandAloneRunner implements RunnerInterface, AutoCloseable {
         if (workerEnabled) {
             // FIXME: For backward-compatibility with Kestra 0.15.x and earliest we still used UUID for Worker ID instead of IdUtils
             String workerID = UUID.randomUUID().toString();
-            Worker worker = applicationContext.createBean(Worker.class, workerID, workerThread, null);
+            Worker worker = applicationContext.createBean(DefaultWorker.class, workerID, workerThread, null);
             applicationContext.registerSingleton(worker); //
             poolExecutor.execute(worker);
             servers.add(worker);
         }
 
         if (schedulerEnabled) {
-            AbstractScheduler scheduler = applicationContext.getBean(AbstractScheduler.class);
+            Scheduler scheduler = applicationContext.getBean(Scheduler.class);
             poolExecutor.execute(scheduler);
             servers.add(scheduler);
         }
 
         if (indexerEnabled) {
-            IndexerInterface indexer = applicationContext.getBean(IndexerInterface.class);
+            Indexer indexer = applicationContext.getBean(Indexer.class);
             poolExecutor.execute(indexer);
             servers.add(indexer);
         }

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

@@ -8,7 +8,7 @@ import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
 import io.kestra.core.runners.FlowInputOutput;
 import io.kestra.core.runners.RunnerUtils;
-import io.kestra.core.runners.StandAloneRunner;
+import io.kestra.cli.StandAloneRunner;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
 import jakarta.validation.ConstraintViolationException;
@@ -72,7 +72,6 @@ public class FlowTestCommand extends AbstractApiCommand {
     public Integer call() throws Exception {
         super.call();
 
-        StandAloneRunner runner = applicationContext.getBean(StandAloneRunner.class);
         LocalFlowRepositoryLoader repositoryLoader = applicationContext.getBean(LocalFlowRepositoryLoader.class);
         FlowRepositoryInterface flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
         FlowInputOutput flowInputOutput = applicationContext.getBean(FlowInputOutput.class);
@@ -89,7 +88,7 @@ public class FlowTestCommand extends AbstractApiCommand {
             inputs.put(this.inputs.get(i), this.inputs.get(i+1));
         }
 
-        try {
+        try (StandAloneRunner runner = applicationContext.createBean(StandAloneRunner.class);){
             runner.run();
             repositoryLoader.load(tenantService.getTenantId(tenantId), file.toFile());
 
@@ -103,8 +102,6 @@ public class FlowTestCommand extends AbstractApiCommand {
                 (flow, execution) -> flowInputOutput.readExecutionInputs(flow, execution, inputs),
                 Duration.ofHours(1)
             );
-
-            runner.close();
         } catch (ConstraintViolationException e) {
             throw new CommandLine.ParameterException(this.spec.commandLine(), e.getMessage());
         } catch (IOException | TimeoutException e) {

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.core.services.SkipExecutionService;
+import io.kestra.executor.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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.runners.IndexerInterface;
+import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
@@ -29,7 +29,7 @@ public class IndexerCommand extends AbstractServerCommand {
     public Integer call() throws Exception {
         super.call();
 
-        IndexerInterface indexer = applicationContext.getBean(IndexerInterface.class);
+        Indexer indexer = applicationContext.getBean(Indexer.class);
         indexer.run();
 
         Await.until(() -> !this.applicationContext.isRunning());

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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.schedulers.AbstractScheduler;
+import io.kestra.scheduler.AbstractScheduler;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;

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

@@ -6,8 +6,8 @@ import io.kestra.cli.services.TenantIdSelectorService;
 import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
-import io.kestra.core.runners.StandAloneRunner;
-import io.kestra.core.services.SkipExecutionService;
+import io.kestra.cli.StandAloneRunner;
+import io.kestra.executor.SkipExecutionService;
 import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
@@ -109,26 +109,27 @@ public class StandAloneCommand extends AbstractServerCommand {
             }
         }
 
-        StandAloneRunner standAloneRunner = applicationContext.getBean(StandAloneRunner.class);
+        try (StandAloneRunner standAloneRunner = applicationContext.getBean(StandAloneRunner.class)) {
 
-        if (this.workerThread == 0) {
-            standAloneRunner.setWorkerEnabled(false);
-        } else {
-            standAloneRunner.setWorkerThread(this.workerThread);
+            if (this.workerThread == 0) {
+                standAloneRunner.setWorkerEnabled(false);
+            } else {
+                standAloneRunner.setWorkerThread(this.workerThread);
+            }
+
+            if (this.indexerDisabled) {
+                standAloneRunner.setIndexerEnabled(false);
+            }
+
+            standAloneRunner.run();
+
+            if (fileWatcher != null) {
+                fileWatcher.startListeningFromConfig();
+            }
+
+            Await.until(() -> !this.applicationContext.isRunning());
         }
 
-        if (this.indexerDisabled) {
-            standAloneRunner.setIndexerEnabled(false);
-        }
-
-        standAloneRunner.run();
-
-        if (fileWatcher != null) {
-            fileWatcher.startListeningFromConfig();
-        }
-
-        Await.until(() -> !this.applicationContext.isRunning());
-
         return 0;
     }
 }

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

@@ -2,7 +2,7 @@ package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.runners.IndexerInterface;
+import io.kestra.core.runners.Indexer;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
 import io.micronaut.context.ApplicationContext;
@@ -54,7 +54,7 @@ public class WebServerCommand extends AbstractServerCommand {
         if (!indexerDisabled) {
             log.info("Starting an embedded indexer, this can be disabled by using `--no-indexer`.");
             poolExecutor = executorsUtils.cachedThreadPool("webserver-indexer");
-            poolExecutor.execute(applicationContext.getBean(IndexerInterface.class));
+            poolExecutor.execute(applicationContext.getBean(Indexer.class));
             shutdownHook(false, () -> poolExecutor.shutdown());
         }
 

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

@@ -10,24 +10,21 @@ import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.services.FlowListenersInterface;
 import io.kestra.core.services.PluginDefaultService;
 import io.micronaut.context.annotation.Requires;
-import io.micronaut.context.annotation.Value;
 import io.micronaut.scheduling.io.watch.FileWatchConfiguration;
 import jakarta.annotation.Nullable;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import jakarta.validation.ConstraintViolationException;
+import java.util.concurrent.CopyOnWriteArrayList;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
 import java.nio.charset.Charset;
 import java.nio.file.*;
 import java.nio.file.attribute.BasicFileAttributes;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 
-import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
-
 @Singleton
 @Slf4j
 @Requires(property = "micronaut.io.watch.enabled", value = "true")
@@ -49,13 +46,9 @@ public class FileChangedEventListener {
     @Inject
     protected FlowListenersInterface flowListeners;
 
-    @Nullable
-    @Value("${micronaut.io.watch.tenantId}")
-    private String tenantId;
-
     FlowFilesManager flowFilesManager;
 
-    private List<FlowWithPath> flows = new ArrayList<>();
+    private List<FlowWithPath> flows = new CopyOnWriteArrayList<>();
 
     private boolean isStarted = false;
 
@@ -113,8 +106,6 @@ public class FileChangedEventListener {
     }
 
     public void startListening(List<Path> paths) throws IOException, InterruptedException {
-        String tenantId = this.tenantId != null ? this.tenantId : MAIN_TENANT;
-
         for (Path path : paths) {
             path.register(watchService, StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_DELETE, StandardWatchEventKinds.ENTRY_MODIFY);
         }
@@ -157,7 +148,7 @@ public class FileChangedEventListener {
                                             flows.add(FlowWithPath.of(flow.get(), filePath.toString()));
                                         }
 
-                                        flowFilesManager.createOrUpdateFlow(GenericFlow.fromYaml(tenantId, content));
+                                        flowFilesManager.createOrUpdateFlow(GenericFlow.fromYaml(getTenantIdFromPath(filePath), content));
                                         log.info("Flow {} from file {} has been created or modified", flow.get().getId(), entry);
                                     }
 
@@ -201,8 +192,6 @@ public class FileChangedEventListener {
     }
 
     private void loadFlowsFromFolder(Path folder) {
-        String tenantId = this.tenantId != null ? this.tenantId : MAIN_TENANT;
-
         try {
             Files.walkFileTree(folder, new SimpleFileVisitor<Path>() {
                 @Override
@@ -222,7 +211,7 @@ public class FileChangedEventListener {
 
                         if (flow.isPresent() && flows.stream().noneMatch(flowWithPath -> flowWithPath.uidWithoutRevision().equals(flow.get().uidWithoutRevision()))) {
                             flows.add(FlowWithPath.of(flow.get(), file.toString()));
-                            flowFilesManager.createOrUpdateFlow(GenericFlow.fromYaml(tenantId, content));
+                            flowFilesManager.createOrUpdateFlow(GenericFlow.fromYaml(getTenantIdFromPath(file), content));
                         }
                     }
                     return FileVisitResult.CONTINUE;
@@ -246,10 +235,8 @@ public class FileChangedEventListener {
     }
 
     private Optional<FlowWithSource> parseFlow(String content, Path entry) {
-        String tenantId = this.tenantId != null ? this.tenantId : MAIN_TENANT;
-
         try {
-            FlowWithSource flow = pluginDefaultService.parseFlowWithAllDefaults(tenantId, content, false);
+            FlowWithSource flow = pluginDefaultService.parseFlowWithAllDefaults(getTenantIdFromPath(entry), content, false);
             modelValidator.validate(flow);
             return Optional.of(flow);
         } catch (ConstraintViolationException | FlowProcessingException e) {
@@ -273,4 +260,10 @@ public class FileChangedEventListener {
     private Path buildPath(FlowInterface flow) {
         return fileWatchConfiguration.getPaths().getFirst().resolve(flow.uidWithoutRevision() + ".yml");
     }
+
+    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

@@ -18,6 +18,10 @@ micronaut:
       root:
         paths: classpath:root
         mapping: /**
+  codec:
+    json:
+      additional-types:
+        - application/scim+json
   server:
     max-request-size: 10GB
     multipart:
@@ -78,8 +82,19 @@ micronaut:
       type: scheduled
       core-pool-size: 1
 
-  # Disable OpenTelemetry metrics by default, users that need it must enable it and configure the collector URL.
   metrics:
+    binders:
+      retry:
+        enabled: true
+      netty:
+        queues:
+          enabled: true
+        bytebuf-allocators:
+          enabled: true
+        channels:
+          enabled: true
+
+    # Disable OpenTelemetry metrics by default, users that need it must enable it and configure the collector URL.
     export:
       otlp:
         enabled: false
@@ -92,6 +107,8 @@ jackson:
   serialization-inclusion: non_null
   deserialization:
     FAIL_ON_UNKNOWN_PROPERTIES: false
+  mapper:
+    ACCEPT_CASE_INSENSITIVE_ENUMS: true
 
 endpoints:
   all:
@@ -100,6 +117,10 @@ endpoints:
     sensitive: false
   health:
     details-visible: ANONYMOUS
+    disk-space:
+      enabled: false
+    discovery-client:
+      enabled: false
   loggers:
     write-sensitive: false
   env:
@@ -133,12 +154,44 @@ kestra:
   tutorial-flows:
     # Automatically loads all tutorial flows at startup.
     enabled: true
+
   retries:
     attempts: 5
     multiplier: 2.0
     delay: 1s
     maxDelay: ""
 
+  server:
+    basic-auth:
+      # These URLs will not be authenticated, by default we open some of the Micronaut default endpoints but not all for security reasons
+      open-urls:
+        - "/ping"
+        - "/api/v1/executions/webhook/"
+
+    preview:
+      initial-rows: 100
+      max-rows: 5000
+
+    # The expected time for this server to complete all its tasks before initiating a graceful shutdown.
+    terminationGracePeriod: 5m
+    workerTaskRestartStrategy: AFTER_TERMINATION_GRACE_PERIOD
+    # Configuration for Liveness and Heartbeat mechanism between servers.
+    liveness:
+      enabled: true
+      # The expected time between liveness probe.
+      interval: 10s
+      # The timeout used to detect service failures.
+      timeout: 1m
+      # The time to wait before executing a liveness probe.
+      initialDelay: 1m
+      # The expected time between service heartbeats.
+      heartbeatInterval: 3s
+    service:
+      purge:
+        initial-delay: 1h
+        fixed-delay: 1d
+        retention: 30d
+
   jdbc:
     queues:
       min-poll-interval: 25ms
@@ -150,7 +203,7 @@ kestra:
       fixed-delay: 1h
       retention: 7d
       types:
-        - type : io.kestra.core.models.executions.LogEntry
+        - type: io.kestra.core.models.executions.LogEntry
           retention: 1h
         - type: io.kestra.core.models.executions.MetricEntry
           retention: 1h
@@ -182,37 +235,12 @@ kestra:
   traces:
     root: DISABLED
 
-  server:
-    basic-auth:
-      # These URLs will not be authenticated, by default we open some of the Micronaut default endpoints but not all for security reasons
-      open-urls:
-        - "/ping"
-        - "/api/v1/executions/webhook/"
-    preview:
-      initial-rows: 100
-      max-rows: 5000
-    # The expected time for this server to complete all its tasks before initiating a graceful shutdown.
-    terminationGracePeriod: 5m
-    workerTaskRestartStrategy: AFTER_TERMINATION_GRACE_PERIOD
-    # Configuration for Liveness and Heartbeat mechanism between servers.
-    liveness:
-      enabled: true
-      # The expected time between liveness probe.
-      interval: 10s
-      # The timeout used to detect service failures.
-      timeout: 1m
-      # The time to wait before executing a liveness probe.
-      initialDelay: 1m
-      # The expected time between service heartbeats.
-      heartbeatInterval: 3s
-    service:
-      purge:
-        initial-delay: 1h
-        fixed-delay: 1d
-        retention: 30d
+  ui-anonymous-usage-report:
+    enabled: true
+
   anonymous-usage-report:
     enabled: true
-    uri: https://api.kestra.io/v1/reports/usages
+    uri: https://api.kestra.io/v1/reports/server-events
     initial-delay: 5m
     fixed-delay: 1h
 

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

@@ -1,13 +1,14 @@
 package io.kestra.cli.services;
 
 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,7 +19,6 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-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;
 
@@ -56,10 +56,11 @@ class FileChangedEventListenerTest {
         }
     }
 
-    @RetryingTest(5) // Flaky on CI but always pass locally
+    @Test
     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 = """
@@ -71,30 +72,33 @@ class FileChangedEventListenerTest {
                 type: io.kestra.plugin.core.log.Log
                 message: Hello World! 🚀
             """;
-        Files.write(Path.of(FILE_WATCH + "/myflow.yaml"), flow.getBytes());
+
+        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");
 
         // delete the flow
-        Files.delete(Path.of(FILE_WATCH + "/myflow.yaml"));
+        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
+    @Test
     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 = """
@@ -110,21 +114,22 @@ class FileChangedEventListenerTest {
                 values:
                   message: Hello World!
             """;
-        Files.write(Path.of(FILE_WATCH + "/plugin-default.yaml"), pluginDefault.getBytes());
+        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");
 
         // delete both files
-        Files.delete(Path.of(FILE_WATCH + "/plugin-default.yaml"));
+        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

@@ -37,15 +37,15 @@ dependencies {
     implementation 'nl.basjes.gitignore:gitignore-reader'
     implementation group: 'dev.failsafe', name: 'failsafe'
     implementation 'com.github.ben-manes.caffeine:caffeine'
-    implementation 'com.github.ksuid:ksuid:1.1.3'
+    implementation 'com.github.ksuid:ksuid:1.1.4'
     api 'org.apache.httpcomponents.client5:httpclient5'
 
     // plugins
     implementation 'org.apache.maven.resolver:maven-resolver-impl'
-    implementation 'org.apache.maven.resolver:maven-resolver-supplier'
+    implementation 'org.apache.maven.resolver:maven-resolver-supplier-mvn3'
     implementation 'org.apache.maven.resolver:maven-resolver-connector-basic'
     implementation 'org.apache.maven.resolver:maven-resolver-transport-file'
-    implementation 'org.apache.maven.resolver:maven-resolver-transport-http'
+    implementation 'org.apache.maven.resolver:maven-resolver-transport-apache'
 
     // scheduler
     implementation group: 'com.cronutils', name: 'cron-utils'
@@ -63,6 +63,10 @@ dependencies {
         exclude group: 'com.fasterxml.jackson.core'
     }
 
+    // micrometer
+    implementation "io.micronaut.micrometer:micronaut-micrometer-observation"
+    implementation 'io.micrometer:micrometer-java21'
+
     // test
     testAnnotationProcessor project(':processor')
     testImplementation project(':tests')
@@ -70,6 +74,9 @@ dependencies {
     testImplementation project(':repository-memory')
     testImplementation project(':runner-memory')
     testImplementation project(':storage-local')
+    testImplementation project(':worker')
+    testImplementation project(':scheduler')
+    testImplementation project(':executor')
 
     testImplementation "io.micronaut:micronaut-http-client"
     testImplementation "io.micronaut:micronaut-http-server-netty"

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

@@ -53,6 +53,8 @@ import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
+import static io.kestra.core.docs.AbstractClassDocumentation.flattenWithoutType;
+import static io.kestra.core.docs.AbstractClassDocumentation.required;
 import static io.kestra.core.serializers.JacksonMapper.MAP_TYPE_REFERENCE;
 
 @Singleton
@@ -92,12 +94,16 @@ public class JsonSchemaGenerator {
     }
 
     public <T> Map<String, Object> schemas(Class<? extends T> cls, boolean arrayOf, List<String> allowedPluginTypes) {
+        return this.schemas(cls, arrayOf, allowedPluginTypes, false);
+    }
+
+    public <T> Map<String, Object> schemas(Class<? extends T> cls, boolean arrayOf, List<String> allowedPluginTypes, boolean withOutputs) {
         SchemaGeneratorConfigBuilder builder = new SchemaGeneratorConfigBuilder(
             SchemaVersion.DRAFT_7,
             OptionPreset.PLAIN_JSON
         );
 
-        this.build(builder, true, allowedPluginTypes);
+        this.build(builder, true, allowedPluginTypes, withOutputs);
 
         SchemaGeneratorConfig schemaGeneratorConfig = builder.build();
 
@@ -249,6 +255,10 @@ public class JsonSchemaGenerator {
     }
 
     protected void build(SchemaGeneratorConfigBuilder builder, boolean draft7, List<String> allowedPluginTypes) {
+        this.build(builder, draft7, allowedPluginTypes, false);
+    }
+
+    protected void build(SchemaGeneratorConfigBuilder builder, boolean draft7, List<String> allowedPluginTypes, boolean withOutputs) {
 //        builder.withObjectMapper(builder.getObjectMapper().configure(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS, false));
         builder
             .with(new JakartaValidationModule(
@@ -430,6 +440,13 @@ public class JsonSchemaGenerator {
                     if (pluginAnnotation.beta()) {
                         collectedTypeAttributes.put("$beta", true);
                     }
+
+                    if (withOutputs) {
+                        Map<String, Object> outputsSchema = this.outputs(null, scope.getType().getErasedType());
+                        collectedTypeAttributes.set("outputs", context.getGeneratorConfig().createObjectNode().pojoNode(
+                            flattenWithoutType(AbstractClassDocumentation.properties(outputsSchema), required(outputsSchema))
+                        ));
+                    }
                 }
 
                 // handle deprecated tasks

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

@@ -36,6 +36,7 @@ public class Plugin {
     private List<PluginElementMetadata> appBlocks;
     private List<PluginElementMetadata> charts;
     private List<PluginElementMetadata> dataFilters;
+    private List<PluginElementMetadata> dataFiltersKPI;
     private List<PluginElementMetadata> logExporters;
     private List<PluginElementMetadata> additionalPlugins;
     private List<PluginSubGroup.PluginCategory> categories;
@@ -96,6 +97,7 @@ public class Plugin {
         plugin.appBlocks = filterAndGetTypeWithMetadata(registeredPlugin.getAppBlocks(), packagePredicate);
         plugin.charts = filterAndGetTypeWithMetadata(registeredPlugin.getCharts(), packagePredicate);
         plugin.dataFilters = filterAndGetTypeWithMetadata(registeredPlugin.getDataFilters(), packagePredicate);
+        plugin.dataFiltersKPI = filterAndGetTypeWithMetadata(registeredPlugin.getDataFiltersKPI(), packagePredicate);
         plugin.logExporters = filterAndGetTypeWithMetadata(registeredPlugin.getLogExporters(), packagePredicate);
         plugin.additionalPlugins = filterAndGetTypeWithMetadata(registeredPlugin.getAdditionalPlugins(), packagePredicate);
 

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/http/client/HttpClient.java

@@ -6,16 +6,24 @@ import io.kestra.core.http.HttpRequest;
 import io.kestra.core.http.HttpResponse;
 import io.kestra.core.http.client.apache.*;
 import io.kestra.core.http.client.configurations.HttpConfiguration;
+import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
+import io.micrometer.common.KeyValues;
+import io.micrometer.core.instrument.binder.httpcomponents.hc5.ApacheHttpClientContext;
+import io.micrometer.core.instrument.binder.httpcomponents.hc5.DefaultApacheHttpClientObservationConvention;
+import io.micrometer.core.instrument.binder.httpcomponents.hc5.ObservationExecChainHandler;
+import io.micrometer.observation.ObservationRegistry;
 import io.micronaut.http.MediaType;
 import jakarta.annotation.Nullable;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hc.client5.http.ContextBuilder;
 import org.apache.hc.client5.http.auth.*;
 import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.ChainElement;
 import org.apache.hc.client5.http.impl.DefaultAuthenticationStrategy;
 import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
@@ -30,7 +38,6 @@ import org.apache.hc.core5.http.io.HttpClientResponseHandler;
 import org.apache.hc.core5.http.io.entity.EntityUtils;
 import org.apache.hc.core5.ssl.SSLContexts;
 import org.apache.hc.core5.util.Timeout;
-import org.codehaus.plexus.util.StringUtils;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -50,11 +57,16 @@ public class HttpClient implements Closeable {
     private transient CloseableHttpClient client;
     private final RunContext runContext;
     private final HttpConfiguration configuration;
+    private ObservationRegistry observationRegistry;
 
     @Builder
     public HttpClient(RunContext runContext, @Nullable HttpConfiguration configuration) throws IllegalVariableEvaluationException {
         this.runContext = runContext;
         this.configuration = configuration == null ? HttpConfiguration.builder().build() : configuration;
+        if (runContext instanceof DefaultRunContext defaultRunContext) {
+            this.observationRegistry = defaultRunContext.getApplicationContext().findBean(ObservationRegistry.class).orElse(null);
+        }
+
         this.client = this.createClient();
     }
 
@@ -67,6 +79,13 @@ public class HttpClient implements Closeable {
             .disableDefaultUserAgent()
             .setUserAgent("Kestra");
 
+        if (observationRegistry != null) {
+            // micrometer, must be placed before the retry strategy (see https://docs.micrometer.io/micrometer/reference/reference/httpcomponents.html#_retry_strategy_considerations)
+            builder.addExecInterceptorAfter(ChainElement.RETRY.name(), "micrometer",
+                new ObservationExecChainHandler(observationRegistry, new CustomApacheHttpClientObservationConvention())
+            );
+        }
+
         // logger
         if (this.configuration.getLogs() != null && this.configuration.getLogs().length > 0) {
             if (ArrayUtils.contains(this.configuration.getLogs(), HttpConfiguration.LoggingType.REQUEST_HEADERS) ||
@@ -297,4 +316,14 @@ public class HttpClient implements Closeable {
             this.client.close();
         }
     }
+
+    public static class CustomApacheHttpClientObservationConvention extends DefaultApacheHttpClientObservationConvention {
+        @Override
+        public KeyValues getLowCardinalityKeyValues(ApacheHttpClientContext context) {
+            return KeyValues.concat(
+                super.getLowCardinalityKeyValues(context),
+                KeyValues.of("type", "core-client")
+            );
+        }
+    }
 }

core/src/main/java/io/kestra/core/metrics/MeterRegistryBinderFactory.java

@@ -0,0 +1,34 @@
+package io.kestra.core.metrics;
+
+import io.micrometer.core.instrument.binder.jvm.JvmThreadDeadlockMetrics;
+import io.micrometer.java21.instrument.binder.jdk.VirtualThreadMetrics;
+import io.micronaut.configuration.metrics.annotation.RequiresMetrics;
+import io.micronaut.context.annotation.Bean;
+import io.micronaut.context.annotation.Factory;
+import io.micronaut.context.annotation.Primary;
+import io.micronaut.context.annotation.Requires;
+import jakarta.inject.Singleton;
+
+import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_BINDERS;
+import static io.micronaut.core.util.StringUtils.FALSE;
+
+@Factory
+@RequiresMetrics
+
+public class MeterRegistryBinderFactory {
+    @Bean
+    @Primary
+    @Singleton
+    @Requires(property = MICRONAUT_METRICS_BINDERS + ".jvm.enabled", notEquals = FALSE)
+    public VirtualThreadMetrics virtualThreadMetrics() {
+        return new VirtualThreadMetrics();
+    }
+
+    @Bean
+    @Primary
+    @Singleton
+    @Requires(property = MICRONAUT_METRICS_BINDERS + ".jvm.enabled", notEquals = FALSE)
+    public JvmThreadDeadlockMetrics threadDeadlockMetricsMetrics() {
+        return new JvmThreadDeadlockMetrics();
+    }
+}

core/src/main/java/io/kestra/core/metrics/MetricRegistry.java

@@ -6,7 +6,6 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.runners.*;
-import io.kestra.core.schedulers.SchedulerExecutionWithTrigger;
 import io.micrometer.core.instrument.*;
 import io.micrometer.core.instrument.binder.MeterBinder;
 import io.micrometer.core.instrument.search.Search;
@@ -395,19 +394,6 @@ public class MetricRegistry {
         return triggerContext.getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, triggerContext.getTenantId());
     }
 
-    /**
-     * Return tags for current {@link SchedulerExecutionWithTrigger}.
-     *
-     * @param schedulerExecutionWithTrigger the current SchedulerExecutionWithTrigger
-     * @return tags to apply to metrics
-     */
-    public String[] tags(SchedulerExecutionWithTrigger schedulerExecutionWithTrigger, String... tags) {
-        return ArrayUtils.addAll(
-            this.tags(schedulerExecutionWithTrigger.getExecution()),
-            tags
-        );
-    }
-
     /**
      * Return tags for current {@link ExecutionKilled}
      *

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/collectors/ServiceUsage.java

@@ -62,6 +62,7 @@ public record ServiceUsage(
 
         List<DailyServiceStatistics> statistics = Arrays
             .stream(ServiceType.values())
+            .filter(it -> !it.equals(ServiceType.INVALID))
             .map(type -> of(from, to, repository, type, interval))
             .toList();
         return new ServiceUsage(statistics);

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

@@ -1,74 +0,0 @@
-package io.kestra.core.models.collectors;
-
-import io.kestra.core.models.ServerType;
-import io.micronaut.core.annotation.Introspected;
-import jakarta.annotation.Nullable;
-import jakarta.validation.Valid;
-import jakarta.validation.constraints.NotNull;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.experimental.SuperBuilder;
-import lombok.extern.jackson.Jacksonized;
-
-import java.time.Instant;
-import java.time.ZoneId;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-@SuperBuilder(toBuilder = true)
-@Getter
-@Jacksonized
-@Introspected
-@AllArgsConstructor
-public class Usage {
-    @NotNull
-    private final String uuid;
-
-    @NotNull
-    private final String startUuid;
-
-    @NotNull
-    private final String instanceUuid;
-
-    @NotNull
-    private final ServerType serverType;
-
-    @NotNull
-    private final String version;
-
-    @NotNull
-    private final ZoneId zoneId;
-
-    @Nullable
-    private final String uri;
-
-    @Nullable
-    private final Set<String> environments;
-
-    @NotNull
-    private final Instant startTime;
-
-    @Valid
-    private final HostUsage host;
-
-    @Valid
-    private final ConfigurationUsage configurations;
-
-    @Valid
-    private final List<PluginUsage> plugins;
-
-    @Valid
-    private final FlowUsage flows;
-
-    @Valid
-    private final ExecutionUsage executions;
-
-    @Valid
-    @Nullable
-    private ServiceUsage services;
-
-    @Valid
-    @Nullable
-    private List<PluginMetric> pluginMetrics;
-}

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

@@ -441,6 +441,28 @@ public class Execution implements DeletedInterface, TenantInterface {
         @Nullable List<ResolvedTask> resolvedErrors,
         @Nullable List<ResolvedTask> resolvedFinally,
         TaskRun parentTaskRun
+    ) {
+        return findTaskDependingFlowState(resolvedTasks, resolvedErrors, resolvedFinally, parentTaskRun, null);
+    }
+
+    /**
+     * Determine if the current execution is on error &amp; normal tasks
+     * <p>
+     * if the current have errors, return tasks from errors if not, return the normal tasks
+     *
+     * @param resolvedTasks normal tasks
+     * @param resolvedErrors errors tasks
+     * @param resolvedFinally finally tasks
+     * @param parentTaskRun the parent task
+     * @param terminalState the parent task terminal state
+     * @return the flow we need to follow
+     */
+    public List<ResolvedTask> findTaskDependingFlowState(
+        List<ResolvedTask> resolvedTasks,
+        @Nullable List<ResolvedTask> resolvedErrors,
+        @Nullable List<ResolvedTask> resolvedFinally,
+        TaskRun parentTaskRun,
+        @Nullable State.Type terminalState
     ) {
         resolvedTasks = removeDisabled(resolvedTasks);
         resolvedErrors = removeDisabled(resolvedErrors);
@@ -454,10 +476,15 @@ public class Execution implements DeletedInterface, TenantInterface {
             return resolvedFinally == null ? Collections.emptyList() : resolvedFinally;
         }
 
-        // Check if flow has failed task
+        // check if the parent task should fail, and there is error tasks so we start them
+        if (errorsFlow.isEmpty() && terminalState == State.Type.FAILED) {
+            return resolvedErrors == null ? resolvedFinally == null ? Collections.emptyList() : resolvedFinally : resolvedErrors;
+        }
+
+        // Check if flow has failed tasks
         if (!errorsFlow.isEmpty() || this.hasFailed(resolvedTasks, parentTaskRun)) {
             // Check if among the failed task, they will be retried
-            if (!this.hasFailedNoRetry(resolvedTasks, parentTaskRun)) {
+            if (!this.hasFailedNoRetry(resolvedTasks, parentTaskRun) && terminalState != State.Type.FAILED) {
                 return Collections.emptyList();
             }
 
@@ -666,6 +693,11 @@ public class Execution implements DeletedInterface, TenantInterface {
 
     public State.Type guessFinalState(List<ResolvedTask> currentTasks, TaskRun parentTaskRun,
         boolean allowFailure, boolean allowWarning) {
+        return guessFinalState(currentTasks, parentTaskRun, allowFailure, allowWarning, State.Type.SUCCESS);
+    }
+
+    public State.Type guessFinalState(List<ResolvedTask> currentTasks, TaskRun parentTaskRun,
+                                      boolean allowFailure, boolean allowWarning, State.Type terminalState) {
         List<TaskRun> taskRuns = this.findTaskRunByTasks(currentTasks, parentTaskRun);
         var state = this
             .findLastByState(taskRuns, State.Type.KILLED)
@@ -682,7 +714,7 @@ public class Execution implements DeletedInterface, TenantInterface {
                 .findLastByState(taskRuns, State.Type.PAUSED)
                 .map(taskRun -> taskRun.getState().getCurrent())
             )
-            .orElse(State.Type.SUCCESS);
+            .orElse(terminalState);
 
         if (state == State.Type.FAILED && allowFailure) {
             if (allowWarning) {
@@ -1040,6 +1072,16 @@ public class Execution implements DeletedInterface, TenantInterface {
         return result;
     }
 
+    /**
+     * Find all children of this {@link TaskRun}.
+     */
+    public List<TaskRun> findChildren(TaskRun parentTaskRun) {
+        return taskRunList.stream()
+            .filter(taskRun -> parentTaskRun.getId().equals(taskRun.getParentTaskRunId()))
+            .toList();
+    }
+
+
     public List<String> findParentsValues(TaskRun taskRun, boolean withCurrent) {
         return (withCurrent ?
             Stream.concat(findParents(taskRun).stream(), Stream.of(taskRun)) :

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

@@ -4,6 +4,8 @@ import com.fasterxml.jackson.annotation.JsonSetter;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.kestra.core.models.flows.input.*;
+import io.kestra.core.models.property.Property;
+import io.kestra.core.runners.RunContext;
 import io.micronaut.core.annotation.Introspected;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.ConstraintViolationException;
@@ -16,6 +18,8 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.function.Function;
+
 @SuppressWarnings("deprecation")
 @SuperBuilder
 @Getter
@@ -78,7 +82,7 @@ public abstract class Input<T> implements Data {
     @Schema(
         title = "The default value to use if no value is specified."
     )
-    T defaults;
+    Property<T> defaults;
 
     @Schema(
         title = "The display name of the input."

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

@@ -43,4 +43,11 @@ public class Output implements Data {
     Type type;
 
     String displayName;
+    
+    /**
+     * Specifies whether the output is required or not.
+     * <p>
+     * By default, an output is always required.
+     */
+    Boolean required;
 }

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

@@ -6,19 +6,21 @@ import jakarta.validation.ConstraintViolationException;
 import jakarta.validation.constraints.NotNull;
 
 /**
- * Represents a
+ * Represents an input along with its associated value and validation state.
  *
- * @param input     The flow's {@link Input}.
- * @param value     The flow's input value/data.
- * @param enabled   Specify whether the input is enabled.
- * @param exception The input validation exception.
+ * @param input     The {@link Input} definition of the flow.
+ * @param value     The provided value for the input.
+ * @param enabled   {@code true} if the input is enabled; {@code false} otherwise.
+ * @param isDefault {@code true} if the provided value is the default; {@code false} otherwise.
+ * @param exception The validation exception, if the input value is invalid; {@code null} otherwise.
  */
 public record InputAndValue(
     Input<?> input,
     Object value,
     boolean enabled,
+    boolean isDefault,
     ConstraintViolationException exception) {
-
+    
     /**
      * Creates a new {@link InputAndValue} instance.
      *
@@ -26,6 +28,6 @@ public record InputAndValue(
      * @param value The value.
      */
     public InputAndValue(@NotNull Input<?> input, @Nullable Object value) {
-        this(input, value, true, null);
+        this(input, value, true, false, null);
     }
 }

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

@@ -3,6 +3,7 @@ package io.kestra.core.models.flows.input;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.RenderableInput;
 import io.kestra.core.models.flows.Type;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.validations.Regex;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -13,6 +14,7 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -56,6 +58,23 @@ public class MultiselectInput extends Input<List<String>> implements ItemTypeInt
     @Builder.Default
     Boolean allowCustomValue = false;
 
+    @Schema(
+        title = "Whether the first value of the multi-select should be selected by default."
+    )
+    @NotNull
+    @Builder.Default
+    Boolean autoSelectFirst = false;
+
+    @Override
+    public Property<List<String>> getDefaults() {
+        Property<List<String>> baseDefaults = super.getDefaults();
+        if (baseDefaults == null && autoSelectFirst && !Optional.ofNullable(values).map(Collection::isEmpty).orElse(true)) {
+            return Property.ofValue(List.of(values.getFirst()));
+        }
+
+        return baseDefaults;
+    }
+
     @Override
     public void validate(List<String> inputs) throws ConstraintViolationException {
         if (values != null && options != null) {
@@ -100,6 +119,7 @@ public class MultiselectInput extends Input<List<String>> implements ItemTypeInt
                 .dependsOn(getDependsOn())
                 .itemType(getItemType())
                 .displayName(getDisplayName())
+                .autoSelectFirst(getAutoSelectFirst())
                 .build();
         }
         return this;

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

@@ -2,6 +2,7 @@ package io.kestra.core.models.flows.input;
 
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.RenderableInput;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.validations.Regex;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -12,6 +13,7 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -46,6 +48,23 @@ public class SelectInput extends Input<String> implements RenderableInput {
     @Builder.Default
     Boolean isRadio = false;
 
+    @Schema(
+        title = "Whether the first value of the select should be selected by default."
+    )
+    @NotNull
+    @Builder.Default
+    Boolean autoSelectFirst = false;
+
+    @Override
+    public Property<String> getDefaults() {
+        Property<String> baseDefaults = super.getDefaults();
+        if (baseDefaults == null && autoSelectFirst && !Optional.ofNullable(values).map(Collection::isEmpty).orElse(true)) {
+            return Property.ofValue(values.getFirst());
+        }
+
+        return baseDefaults;
+    }
+
     @Override
     public void validate(String input) throws ConstraintViolationException {
         if (!values.contains(input) && this.getRequired()) {
@@ -78,6 +97,7 @@ public class SelectInput extends Input<String> implements RenderableInput {
                 .dependsOn(getDependsOn())
                 .displayName(getDisplayName())
                 .isRadio(getIsRadio())
+                .autoSelectFirst(getAutoSelectFirst())
                 .build();
         }
         return this;

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

@@ -68,6 +68,19 @@ public class Property<T> {
     String getExpression() {
         return expression;
     }
+    
+    /**
+     * Returns a new {@link Property} with no cached rendered value,
+     * so that the next render will evaluate its original Pebble expression.
+     * <p>
+     * The returned property will still cache its rendered result.
+     * To re-evaluate on a subsequent render, call {@code skipCache()} again.
+     *
+     * @return a new {@link Property} without a pre-rendered value
+     */
+    public Property<T> skipCache() {
+        return Property.ofExpression(expression);
+    }
 
     /**
      * Build a new Property object with a value already set.<br>
@@ -132,8 +145,8 @@ public class Property<T> {
      *
      * @see io.kestra.core.runners.RunContextProperty#as(Class)
      */
-    public static <T> T as(Property<T> property, RunContext runContext, Class<T> clazz) throws IllegalVariableEvaluationException {
-        return as(property, runContext, clazz, Map.of());
+    public static <T> T as(Property<T> property, PropertyContext context, Class<T> clazz) throws IllegalVariableEvaluationException {
+        return as(property, context, clazz, Map.of());
     }
 
     /**
@@ -143,9 +156,9 @@ public class Property<T> {
      *
      * @see io.kestra.core.runners.RunContextProperty#as(Class, Map)
      */
-    public static <T> T as(Property<T> property, RunContext runContext, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
+    public static <T> T as(Property<T> property, PropertyContext context, Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
         if (property.value == null) {
-            String rendered =  runContext.render(property.expression, variables);
+            String rendered =  context.render(property.expression, variables);
             property.value = MAPPER.convertValue(rendered, clazz);
         }
 
@@ -159,8 +172,8 @@ public class Property<T> {
      *
      * @see io.kestra.core.runners.RunContextProperty#asList(Class)
      */
-    public static <T, I> T asList(Property<T> property, RunContext runContext, Class<I> itemClazz) throws IllegalVariableEvaluationException {
-        return asList(property, runContext, itemClazz, Map.of());
+    public static <T, I> T asList(Property<T> property, PropertyContext context, Class<I> itemClazz) throws IllegalVariableEvaluationException {
+        return asList(property, context, itemClazz, Map.of());
     }
 
     /**
@@ -171,7 +184,7 @@ public class Property<T> {
      * @see io.kestra.core.runners.RunContextProperty#asList(Class, Map)
      */
     @SuppressWarnings("unchecked")
-    public static <T, I> T asList(Property<T> property, RunContext runContext, Class<I> itemClazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
+    public static <T, I> T asList(Property<T> property, PropertyContext context, Class<I> itemClazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
         if (property.value == null) {
             JavaType type = MAPPER.getTypeFactory().constructCollectionLikeType(List.class, itemClazz);
             try {
@@ -179,7 +192,7 @@ public class Property<T> {
                 // We need to detect if the expression is already a list or if it's a pebble expression (for eg. referencing a variable containing a list).
                 // Doing that allows us to, if it's an expression, first render then read it as a list.
                 if (trimmedExpression.startsWith("{{") && trimmedExpression.endsWith("}}")) {
-                    property.value = MAPPER.readValue(runContext.render(property.expression, variables), type);
+                    property.value = MAPPER.readValue(context.render(property.expression, variables), type);
                 }
                 // Otherwise, if it's already a list, we read it as a list first then render it from run context which handle list rendering by rendering each item of the list
                 else {
@@ -187,9 +200,9 @@ public class Property<T> {
                     property.value = (T) asRawList.stream()
                         .map(throwFunction(item -> {
                             if (item instanceof String str) {
-                                return MAPPER.convertValue(runContext.render(str, variables), itemClazz);
+                                return MAPPER.convertValue(context.render(str, variables), itemClazz);
                             } else if (item instanceof Map map) {
-                                return MAPPER.convertValue(runContext.render(map, variables), itemClazz);
+                                return MAPPER.convertValue(context.render(map, variables), itemClazz);
                             }
                             return item;
                         }))

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

@@ -0,0 +1,38 @@
+package io.kestra.core.models.property;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.runners.VariableRenderer;
+
+import java.util.Map;
+
+/**
+ * Contextual object for rendering properties.
+ * 
+ * @see Property
+ */
+public interface PropertyContext {
+    
+    String render(String inline, Map<String, Object> variables) throws IllegalVariableEvaluationException;
+    
+    Map<String, Object> render(Map<String, Object> inline, Map<String, Object> variables) throws IllegalVariableEvaluationException;
+    
+    /**
+     * Static helper method for creating a new {@link PropertyContext} from a given {@link VariableRenderer}.
+     *
+     * @param renderer the {@link VariableRenderer}.
+     * @return a new {@link PropertyContext}.
+     */
+    static PropertyContext create(final VariableRenderer renderer) {
+        return new PropertyContext() {
+            @Override
+            public String render(String inline, Map<String, Object> variables) throws IllegalVariableEvaluationException {
+                return renderer.render(inline, variables);
+            }
+            
+            @Override
+            public Map<String, Object> render(Map<String, Object> inline, Map<String, Object> variables) throws IllegalVariableEvaluationException {
+                return renderer.render(inline, variables);
+            }
+        };
+    }
+}

core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleCondition.java

@@ -3,7 +3,6 @@ package io.kestra.core.models.triggers.multipleflows;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
-import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.triggers.TimeWindow;
 import io.kestra.core.utils.Rethrow;
 import org.slf4j.Logger;
@@ -24,7 +23,7 @@ public interface MultipleCondition extends Rethrow.PredicateChecked<ConditionCon
 
     /**
      * This conditions will only validate previously calculated value on
-     * {@link io.kestra.core.services.FlowTriggerService#computeExecutionsFromFlowTriggers(Execution, List, Optional)}} and {@link MultipleConditionStorageInterface#save(List)} by the executor.
+     * io.kestra.executor.FlowTriggerService#computeExecutionsFromFlowTriggers(Execution, List, Optional) and {@link MultipleConditionStorageInterface#save(List)} by the executor.
      * The real validation is done here.
      */
     @Override

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

@@ -43,7 +43,7 @@ public class DefaultPluginRegistry implements PluginRegistry {
         static final DefaultPluginRegistry INSTANCE = new DefaultPluginRegistry();
     }
 
-    private final Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> pluginClassByIdentifier = new ConcurrentHashMap<>();
+    protected final Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> pluginClassByIdentifier = new ConcurrentHashMap<>();
     private final Map<PluginBundleIdentifier, RegisteredPlugin> plugins = new ConcurrentHashMap<>();
     private final PluginScanner scanner = new PluginScanner(DefaultPluginRegistry.class.getClassLoader());
     private final AtomicBoolean initialized = new AtomicBoolean(false);
@@ -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());
         }
@@ -103,11 +103,13 @@ public class DefaultPluginRegistry implements PluginRegistry {
      */
     @Override
     public void registerIfAbsent(final Path pluginPath) {
+        long start = System.currentTimeMillis();
         if (isPluginPathValid(pluginPath) && !isPluginPathScanned(pluginPath)) {
             List<RegisteredPlugin> scanned = scanner.scan(pluginPath);
             scanned.forEach(this::register);
             scannedPluginPaths.add(pluginPath);
         }
+        log.debug("Registered if absent plugins from path {} in {} ms", pluginPath, System.currentTimeMillis() - start);
     }
 
     private boolean isPluginPathScanned(final Path pluginPath) {
@@ -119,10 +121,12 @@ public class DefaultPluginRegistry implements PluginRegistry {
      */
     @Override
     public void register(final Path pluginPath) {
+        long start = System.currentTimeMillis();
         if (isPluginPathValid(pluginPath)) {
             List<RegisteredPlugin> scanned = scanner.scan(pluginPath);
             scanned.forEach(this::register);
         }
+        log.debug("Registered plugins from path {} in {} ms", pluginPath, System.currentTimeMillis() - start);
     }
 
     /**
@@ -191,21 +195,28 @@ public class DefaultPluginRegistry implements PluginRegistry {
      */
     public void register(final RegisteredPlugin plugin) {
         final PluginBundleIdentifier identifier = PluginBundleIdentifier.of(plugin);
-
-        // Skip registration if plugin-bundle already exists in the registry.
-        if (containsPluginBundle(identifier)) {
-            return;
+        // Skip registration if the same plugin already exists in the registry.
+        final RegisteredPlugin existing = plugins.get(identifier);
+        if (existing != null && existing.crc32() == plugin.crc32()) {
+            return; // same plugin already registered
         }
 
         lock.lock();
         try {
+            if (existing != null) {
+                unregister(List.of(existing));
+            }
             plugins.put(PluginBundleIdentifier.of(plugin), plugin);
-            pluginClassByIdentifier.putAll(getPluginClassesByIdentifier(plugin));
+            registerAll(getPluginClassesByIdentifier(plugin));
         } finally {
             lock.unlock();
         }
     }
 
+    protected void registerAll(Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> plugins) {
+        pluginClassByIdentifier.putAll(plugins);
+    }
+
     @SuppressWarnings("unchecked")
     protected Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> getPluginClassesByIdentifier(final RegisteredPlugin plugin) {
         Map<PluginIdentifier, PluginClassAndMetadata<? extends Plugin>> classes = new HashMap<>();

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,4 +20,59 @@ import java.net.URL;
 public class ExternalPlugin {
     private final URL location;
     private final URL[] resources;
+    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/PluginClassLoader.java

@@ -46,6 +46,7 @@ public class PluginClassLoader extends URLClassLoader {
         + "|dev.failsafe"
         + "|reactor"
         + "|io.opentelemetry"
+        + "|io.netty"
         + ")\\..*$");
 
     private final ClassLoader parent;

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

@@ -2,10 +2,14 @@ package io.kestra.core.plugins;
 
 import io.kestra.core.models.Plugin;
 
+import java.net.URL;
+import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Predicate;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
 
 /**
  * Registry for managing all Kestra's {@link Plugin}.
@@ -123,4 +127,24 @@ public interface PluginRegistry {
      * @return {@code true} if supported. Otherwise {@code false}.
      */
     boolean isVersioningSupported();
+    
+    /**
+     * Computes a CRC32 hash value representing the current content of the plugin registry.
+     *
+     * @return a {@code long} containing the CRC32 checksum value, serving as a compact
+     *         representation of the registry's content
+     */
+    default long hash() {
+        Checksum crc32 = new CRC32();
+        
+        for (RegisteredPlugin plugin : plugins()) {
+            Optional.ofNullable(plugin.getExternalPlugin())
+                .map(ExternalPlugin::getCrc32)
+                .ifPresent(checksum -> {
+                    byte[] bytes = ByteBuffer.allocate(Long.BYTES).putLong(checksum).array();
+                    crc32.update(bytes, 0, bytes.length);
+                });
+        }
+        return crc32.getValue();
+    }
 }

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

@@ -5,11 +5,15 @@ import lombok.extern.slf4j.Slf4j;
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.nio.ByteBuffer;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.InvalidPathException;
 import java.nio.file.Path;
 import java.util.*;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+import java.util.zip.CRC32;
 
 @Slf4j
 public class PluginResolver {
@@ -119,4 +123,5 @@ public class PluginResolver {
 
         return urls;
     }
+
 }

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

@@ -308,6 +308,10 @@ public class RegisteredPlugin {
         }
         return null;
     }
+    
+    public long crc32() {
+        return Optional.ofNullable(externalPlugin).map(ExternalPlugin::getCrc32).orElse(-1L);
+    }
 
     @Override
     public String toString() {

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

@@ -5,11 +5,9 @@ import io.kestra.core.models.executions.ExecutionKilled;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.executions.MetricEntry;
 import io.kestra.core.models.flows.FlowInterface;
-import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.templates.Template;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.runners.*;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.templates.Template;
 
 public interface QueueFactoryInterface {
     String EXECUTION_NAMED = "executionQueue";
@@ -34,7 +32,7 @@ public interface QueueFactoryInterface {
 
     QueueInterface<Executor> executor();
 
-    QueueInterface<WorkerJob> workerJob();
+    WorkerJobQueueInterface workerJob();
 
     QueueInterface<WorkerTaskResult> workerTaskResult();
 

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

@@ -5,6 +5,7 @@ import io.kestra.core.models.Pauseable;
 import io.kestra.core.utils.Either;
 
 import java.io.Closeable;
+import java.util.List;
 import java.util.function.Consumer;
 
 public interface QueueInterface<T> extends Closeable, Pauseable {
@@ -18,7 +19,15 @@ public interface QueueInterface<T> extends Closeable, Pauseable {
         emitAsync(null, message);
     }
 
-    void emitAsync(String consumerGroup, T message) throws QueueException;
+    default void emitAsync(String consumerGroup, T message) throws QueueException {
+        emitAsync(consumerGroup, List.of(message));
+    }
+
+    default void emitAsync(List<T> messages) throws QueueException {
+        emitAsync(null, messages);
+    }
+
+    void emitAsync(String consumerGroup, List<T> messages) throws QueueException;
 
     default void delete(T message) throws QueueException {
         delete(null, message);

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

@@ -0,0 +1,12 @@
+package io.kestra.core.queues;
+
+import io.kestra.core.exceptions.DeserializationException;
+import io.kestra.core.runners.WorkerJob;
+import io.kestra.core.utils.Either;
+
+import java.util.function.Consumer;
+
+public interface WorkerJobQueueInterface extends QueueInterface<WorkerJob> {
+    
+    Runnable subscribe(String workerId, String workerGroup, Consumer<Either<WorkerJob, DeserializationException>> consumer);
+}

core/src/main/java/io/kestra/core/reporter/AbstractReportable.java

@@ -0,0 +1,29 @@
+package io.kestra.core.reporter;
+
+public abstract class AbstractReportable<T extends Reportable.Event> implements Reportable<T> {
+    
+    private final Type type;
+    private final ReportingSchedule schedule;
+    private final boolean isTenantSupported;
+    
+    public AbstractReportable(Type type, ReportingSchedule schedule, boolean isTenantSupported) {
+        this.type = type;
+        this.schedule = schedule;
+        this.isTenantSupported = isTenantSupported;
+    }
+    
+    @Override
+    public boolean isTenantSupported() {
+        return isTenantSupported;
+    }
+    
+    @Override
+    public Type type() {
+        return type;
+    }
+    
+    @Override
+    public ReportingSchedule schedule() {
+        return schedule;
+    }
+}

core/src/main/java/io/kestra/core/reporter/Reportable.java

@@ -0,0 +1,94 @@
+package io.kestra.core.reporter;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+
+/**
+ * Interface for reporting server event for a specific type.
+ *
+ * @param <T>
+ */
+public interface Reportable<T extends Reportable.Event> {
+    
+    /**
+     * Gets the type of the event to report.
+     */
+    Type type();
+    
+    /**
+     * Gets the reporting schedule.
+     */
+    ReportingSchedule schedule();
+    
+    /**
+     * Generates a report for the given timestamp.
+     *
+     * @param now the time when the report is triggered.
+     * @return an Optional containing the report data if available.
+     */
+    T report(Instant now, TimeInterval interval);
+    
+    default T report(Instant now) {
+        ZonedDateTime to = now.atZone(ZoneId.systemDefault());
+        ZonedDateTime from = to.minus(Duration.ofDays(1));
+        return report(now, new TimeInterval(from, to));
+    }
+    
+    /**
+     * Checks whether this reportable is enabled for scheduled reporting.
+     */
+    boolean isEnabled();
+    
+    /**
+     * Generates a report for the given timestamp and tenant.
+     *
+     * @param now    the time when the report is triggered.
+     * @param tenant the tenant for which the report is triggered.
+     * @return the event to report.
+     */
+    default T report(Instant now, TimeInterval interval, String tenant) {
+        throw new UnsupportedOperationException();
+    }
+    
+    default T report(Instant now, String tenant) {
+        ZonedDateTime to = now.atZone(ZoneId.systemDefault());
+        ZonedDateTime from = to.minus(Duration.ofDays(1));
+        return report(now, new TimeInterval(from, to), tenant);
+    }
+    
+    /**
+     * Checks whether this {@link Reportable} can accept a tenant.
+     * 
+     * @return {@code true} a {@link #report(Instant, TimeInterval, String)} can called, Otherwise {@code false}.
+     */
+    default boolean isTenantSupported() {
+        return false;
+    }
+    
+    record TimeInterval(ZonedDateTime from, ZonedDateTime to){
+        public static TimeInterval of(ZonedDateTime from, ZonedDateTime to) {
+            return new TimeInterval(from, to);
+        }
+    }
+    
+    /**
+     * Marker interface indicating that the returned event 
+     * must be a structured, domain-specific object 
+     * (not a primitive wrapper, String, collection, or other basic type).
+     */
+    interface Event {
+        
+    }
+    
+    /**
+     * Defines the schedule for a report.
+     */
+    interface ReportingSchedule {
+        /**
+         * Determines whether a report should run at the given instant.
+         */
+        boolean shouldRun(Instant now);
+    }
+}

core/src/main/java/io/kestra/core/reporter/ReportableRegistry.java

@@ -0,0 +1,40 @@
+package io.kestra.core.reporter;
+
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Singleton
+@Slf4j
+public class ReportableRegistry {
+    
+    private final Map<Type, Reportable<?>> reportables = new ConcurrentHashMap<>();
+    
+    /**
+     * Creates a new {@link ReportableRegistry} instance.
+     *
+     * @param reportables The {@link Reportable reportables}
+     */
+    @Inject
+    public ReportableRegistry(final List<Reportable<?>> reportables) {
+        reportables.forEach(reportable -> this.reportables.put(reportable.type(), reportable));
+    }
+    
+    public void register(final Reportable<?> reportable) {
+        Objects.requireNonNull(reportable, "reportable must not be null");
+        if (reportables.containsKey(reportable.type())) {
+            log.warn("Event already registered for type '{}'", reportable.type());
+        } else {
+            reportables.put(reportable.type(), reportable);
+        }
+    }
+    
+    public List<Reportable<?>> getAll() {
+        return List.copyOf(reportables.values());
+    }
+}

core/src/main/java/io/kestra/core/reporter/ReportableScheduler.java

@@ -0,0 +1,43 @@
+package io.kestra.core.reporter;
+
+import io.micronaut.context.annotation.Requires;
+import io.micronaut.scheduling.annotation.Scheduled;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Clock;
+import java.time.Instant;
+
+@Singleton
+@Requires(property = "kestra.anonymous-usage-report.enabled", value = "true")
+@Requires(property = "kestra.server-type")
+@Slf4j
+public class ReportableScheduler {
+    
+    private final ReportableRegistry registry;
+    private final ServerEventSender sender;
+    private final Clock clock;
+    
+    @Inject
+    public ReportableScheduler(ReportableRegistry registry, ServerEventSender sender) {
+        this.registry = registry;
+        this.sender = sender;
+        this.clock = Clock.systemDefaultZone();
+    }
+    
+    @Scheduled(fixedDelay = "5m", initialDelay = "${kestra.anonymous-usage-report.initial-delay}")
+    public void tick() {
+        Instant now = clock.instant();
+        for (Reportable<?> r : registry.getAll()) {
+            if (r.isEnabled() && r.schedule().shouldRun(now)) {
+                try {
+                    Object value = r.report(now);
+                    if (value != null) sender.send(now, r.type(), value);
+                } catch (Exception e) {
+                    log.debug("Failed to send report for event-type '{}'", r.type(), e);
+                }
+            }
+        }
+    }
+}

core/src/main/java/io/kestra/core/reporter/Schedules.java

@@ -0,0 +1,57 @@
+package io.kestra.core.reporter;
+
+import io.kestra.core.reporter.Reportable.ReportingSchedule;
+
+import java.time.Duration;
+import java.time.Instant;
+
+/**
+ * Utility class providing common implementations of {@link Reportable.ReportingSchedule}.
+ */
+public class Schedules {
+    
+    /**
+     * Creates a reporting schedule that triggers after the specified period has elapsed
+     * since the last execution.
+     *
+     * @param period the duration between successive runs; must be positive
+     * @return a {@link Reportable.ReportingSchedule} that runs at the given interval
+     * @throws IllegalArgumentException if {@code period} is zero or negative
+     */
+    public static ReportingSchedule every(final Duration period) {
+        if (period.isZero() || period.isNegative()) {
+            throw new IllegalArgumentException("Period must be positive");
+        }
+        
+        return new ReportingSchedule() {
+            private Instant lastRun = Instant.EPOCH;
+            
+            @Override
+            public boolean shouldRun(Instant now) {
+                if (Duration.between(lastRun, now).compareTo(period) >= 0) {
+                    lastRun = now;
+                    return true;
+                }
+                return false;
+            }
+        };
+    }
+    
+    /**
+     * Creates a reporting schedule that triggers once every hour.
+     *
+     * @return a schedule running every 1 hour
+     */
+    public static ReportingSchedule hourly() {
+        return every(Duration.ofHours(1));
+    }
+    
+    /**
+     * Creates a reporting schedule that triggers once every day.
+     *
+     * @return a schedule running every 24 hours
+     */
+    public static ReportingSchedule daily() {
+        return every(Duration.ofDays(1));
+    }
+}

core/src/main/java/io/kestra/core/reporter/ServerEvent.java

@@ -0,0 +1,31 @@
+package io.kestra.core.reporter;
+
+import com.fasterxml.jackson.annotation.JsonUnwrapped;
+import io.kestra.core.models.ServerType;
+import lombok.Builder;
+
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+
+/**
+ * Represents a Kestra Server Event.
+ */
+@Builder(toBuilder = true)
+public record ServerEvent(
+    String instanceUuid,
+    String sessionUuid,
+    ServerType serverType,
+    String serverVersion,
+    ZoneId zoneId,
+    Object payload,
+    String uuid,
+    ZonedDateTime reportedAt
+) {
+    
+    @JsonUnwrapped
+    public Object payload() {
+        return payload;
+    }
+    
+}
+

core/src/main/java/io/kestra/core/reporter/ServerEventSender.java

@@ -0,0 +1,91 @@
+package io.kestra.core.reporter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.models.ServerType;
+import io.kestra.core.models.collectors.Result;
+import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.services.InstanceService;
+import io.kestra.core.utils.IdUtils;
+import io.kestra.core.utils.VersionProvider;
+import io.micronaut.context.annotation.Value;
+import io.micronaut.core.type.Argument;
+import io.micronaut.http.HttpRequest;
+import io.micronaut.http.MutableHttpRequest;
+import io.micronaut.http.client.annotation.Client;
+import io.micronaut.http.client.exceptions.HttpClientResponseException;
+import io.micronaut.http.hateoas.JsonError;
+import io.micronaut.reactor.http.client.ReactorHttpClient;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+
+import java.net.URI;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.UUID;
+
+@Singleton
+@Slf4j
+public class ServerEventSender {
+    
+    private static final String SESSION_UUID = IdUtils.create();
+    private static final ObjectMapper OBJECT_MAPPER = JacksonMapper.ofJson();
+    
+    @Inject
+    @Client
+    private ReactorHttpClient client;
+    
+    @Inject
+    private VersionProvider versionProvider;
+    
+    @Inject
+    private InstanceService instanceService;
+    
+    private final ServerType serverType;
+    
+    @Value("${kestra.anonymous-usage-report.uri}")
+    protected URI url;
+    
+    public ServerEventSender( ) {
+        this.serverType = KestraContext.getContext().getServerType();
+    }
+    
+    public void send(final Instant now, final Type type, Object event) {
+        ServerEvent serverEvent = ServerEvent
+            .builder()
+            .uuid(UUID.randomUUID().toString())
+            .sessionUuid(SESSION_UUID)
+            .instanceUuid(instanceService.fetch())
+            .serverType(serverType)
+            .serverVersion(versionProvider.getVersion())
+            .reportedAt(now.atZone(ZoneId.systemDefault()))
+            .payload(event)
+            .zoneId(ZoneId.systemDefault())
+            .build();
+        try {
+            MutableHttpRequest<ServerEvent> request = this.request(serverEvent, type);
+            
+            if (log.isTraceEnabled()) {
+                log.trace("Report anonymous usage: '{}'", OBJECT_MAPPER.writeValueAsString(serverEvent));
+            }
+            
+            this.handleResponse(client.toBlocking().retrieve(request, Argument.of(Result.class), Argument.of(JsonError.class)));
+        } catch (HttpClientResponseException t) {
+            log.trace("Unable to report anonymous usage with body '{}'", t.getResponse().getBody(String.class), t);
+        } catch (Exception t) {
+            log.trace("Unable to handle anonymous usage", t);
+        }
+    }
+    
+    private void handleResponse (Result result){
+        
+    }
+    
+    protected MutableHttpRequest<ServerEvent> request(ServerEvent event, Type type) throws Exception {
+        URI baseUri = URI.create(this.url.toString().endsWith("/") ? this.url.toString() : this.url + "/");
+        URI resolvedUri = baseUri.resolve(type.name().toLowerCase());
+        return HttpRequest.POST(resolvedUri, event)
+            .header("User-Agent", "Kestra/" + versionProvider.getVersion());
+    }
+}

core/src/main/java/io/kestra/core/reporter/Type.java

@@ -0,0 +1,9 @@
+package io.kestra.core.reporter;
+
+/**
+ * A reportable event type.
+ */
+public interface Type {
+    
+    String name();
+}

core/src/main/java/io/kestra/core/reporter/Types.java

@@ -0,0 +1,12 @@
+package io.kestra.core.reporter;
+
+/**
+ * All supported reportable event type.
+ */
+public enum Types implements Type {
+    USAGE,
+    SYSTEM_INFORMATION,
+    PLUGIN_METRICS,
+    SERVICE_USAGE,
+    PLUGIN_USAGE;
+}

core/src/main/java/io/kestra/core/reporter/model/Count.java

@@ -0,0 +1,6 @@
+package io.kestra.core.reporter.model;
+
+public record Count(
+    long count
+) {
+}

core/src/main/java/io/kestra/core/reporter/reports/FeatureUsageReport.java

@@ -0,0 +1,80 @@
+package io.kestra.core.reporter.reports;
+
+import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.models.ServerType;
+import io.kestra.core.models.collectors.ExecutionUsage;
+import io.kestra.core.models.collectors.FlowUsage;
+import io.kestra.core.reporter.AbstractReportable;
+import io.kestra.core.reporter.Schedules;
+import io.kestra.core.reporter.Types;
+import io.kestra.core.reporter.model.Count;
+import io.kestra.core.repositories.DashboardRepositoryInterface;
+import io.kestra.core.repositories.ExecutionRepositoryInterface;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.micronaut.core.annotation.Introspected;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.Getter;
+import lombok.experimental.SuperBuilder;
+import lombok.extern.jackson.Jacksonized;
+
+import java.time.Instant;
+import java.util.Objects;
+
+@Singleton
+public class FeatureUsageReport extends AbstractReportable<FeatureUsageReport.UsageEvent> {
+    
+    private final FlowRepositoryInterface flowRepository;
+    private final ExecutionRepositoryInterface executionRepository;
+    private final DashboardRepositoryInterface dashboardRepository;
+    private final boolean enabled;
+    
+    @Inject
+    public FeatureUsageReport(FlowRepositoryInterface flowRepository,
+                              ExecutionRepositoryInterface executionRepository,
+                              DashboardRepositoryInterface dashboardRepository) {
+        super(Types.USAGE, Schedules.hourly(), true);
+        this.flowRepository = flowRepository;
+        this.executionRepository = executionRepository;
+        this.dashboardRepository = dashboardRepository;
+        
+        ServerType serverType = KestraContext.getContext().getServerType();
+        this.enabled = ServerType.EXECUTOR.equals(serverType) || ServerType.STANDALONE.equals(serverType);
+    }
+    
+    @Override
+    public UsageEvent report(final Instant now, TimeInterval interval) {
+        return UsageEvent
+            .builder()
+            .flows(FlowUsage.of(flowRepository))
+            .executions(ExecutionUsage.of(executionRepository, interval.from(), interval.to()))
+            .dashboards(new Count(dashboardRepository.count()))
+            .build();
+    }
+    
+    @Override
+    public boolean isEnabled() {
+        return enabled;
+    }
+    
+    @Override
+    public UsageEvent report(Instant now, TimeInterval interval, String tenant) {
+        Objects.requireNonNull(tenant, "tenant is null");
+        Objects.requireNonNull(interval, "interval is null");
+        return UsageEvent
+            .builder()
+            .flows(FlowUsage.of(tenant, flowRepository))
+            .executions(ExecutionUsage.of(tenant, executionRepository, interval.from(), interval.to()))
+            .build();
+    }
+    
+    @SuperBuilder(toBuilder = true)
+    @Getter
+    @Jacksonized
+    @Introspected
+    public static class UsageEvent implements Event {
+        private ExecutionUsage executions;
+        private FlowUsage flows;
+        private Count dashboards;
+    }
+}

core/src/main/java/io/kestra/core/reporter/reports/PluginMetricReport.java

@@ -0,0 +1,105 @@
+package io.kestra.core.reporter.reports;
+
+import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.ServerType;
+import io.kestra.core.models.collectors.PluginMetric;
+import io.kestra.core.plugins.PluginRegistry;
+import io.kestra.core.reporter.AbstractReportable;
+import io.kestra.core.reporter.Schedules;
+import io.kestra.core.reporter.Types;
+import io.kestra.core.utils.ListUtils;
+import io.micrometer.core.instrument.Timer;
+import io.micronaut.core.annotation.Introspected;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.Builder;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+@Singleton
+public class PluginMetricReport extends AbstractReportable<PluginMetricReport.PluginMetricEvent> {
+    
+    private final PluginRegistry pluginRegistry;
+    private final MetricRegistry metricRegistry;
+    private final boolean enabled;
+    
+    @Inject
+    public PluginMetricReport(PluginRegistry pluginRegistry,
+                              MetricRegistry metricRegistry) {
+        super(Types.PLUGIN_METRICS, Schedules.daily(), false);
+        this.metricRegistry = metricRegistry;
+        this.pluginRegistry = pluginRegistry;
+        
+        ServerType serverType = KestraContext.getContext().getServerType();
+        this.enabled = ServerType.SCHEDULER.equals(serverType) || ServerType.WORKER.equals(serverType) || ServerType.STANDALONE.equals(serverType);
+    }
+    
+    @Override
+    public PluginMetricEvent report(final Instant now, final TimeInterval period) {
+        return PluginMetricEvent
+            .builder()
+            .pluginMetrics(pluginMetrics())
+            .build();
+    }
+    
+    @Override
+    public boolean isEnabled() {
+        return enabled;
+    }
+    
+    @Builder
+    @Introspected
+    public record PluginMetricEvent (
+        List<PluginMetric> pluginMetrics
+    ) implements Event {
+    }
+    
+    private List<PluginMetric> pluginMetrics() {
+        List<PluginMetric> taskMetrics = pluginRegistry.plugins().stream()
+            .flatMap(registeredPlugin -> registeredPlugin.getTasks().stream())
+            .map(Class::getName)
+            .map(this::taskMetric)
+            .flatMap(Optional::stream)
+            .toList();
+        
+        List<PluginMetric> triggerMetrics = pluginRegistry.plugins().stream()
+            .flatMap(registeredPlugin -> registeredPlugin.getTriggers().stream())
+            .map(Class::getName)
+            .map(this::triggerMetric)
+            .flatMap(Optional::stream)
+            .toList();
+        
+        return ListUtils.concat(taskMetrics, triggerMetrics);
+    }
+    
+    private Optional<PluginMetric> taskMetric(String type) {
+        Timer duration = metricRegistry.find(MetricRegistry.METRIC_WORKER_ENDED_DURATION).tag(MetricRegistry.TAG_TASK_TYPE, type).timer();
+        return fromTimer(type, duration);
+    }
+    
+    private Optional<PluginMetric> triggerMetric(String type) {
+        Timer duration = metricRegistry.find(MetricRegistry.METRIC_WORKER_TRIGGER_DURATION).tag(MetricRegistry.TAG_TRIGGER_TYPE, type).timer();
+        
+        if (duration == null) {
+            // this may be because this is a trigger executed by the scheduler, we search there instead
+            duration = metricRegistry.find(MetricRegistry.METRIC_SCHEDULER_TRIGGER_EVALUATION_DURATION).tag(MetricRegistry.TAG_TRIGGER_TYPE, type).timer();
+        }
+        return fromTimer(type, duration);
+    }
+    
+    private Optional<PluginMetric> fromTimer(String type, Timer timer) {
+        if (timer == null || timer.count() == 0) {
+            return Optional.empty();
+        }
+        
+        double count = timer.count();
+        double totalTime = timer.totalTime(TimeUnit.MILLISECONDS);
+        double meanTime = timer.mean(TimeUnit.MILLISECONDS);
+        
+        return Optional.of(new PluginMetric(type, count, totalTime, meanTime));
+    }
+}

core/src/main/java/io/kestra/core/reporter/reports/PluginUsageReport.java

@@ -0,0 +1,51 @@
+package io.kestra.core.reporter.reports;
+
+import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.models.ServerType;
+import io.kestra.core.models.collectors.PluginUsage;
+import io.kestra.core.plugins.PluginRegistry;
+import io.kestra.core.reporter.AbstractReportable;
+import io.kestra.core.reporter.Schedules;
+import io.kestra.core.reporter.Types;
+import io.micronaut.core.annotation.Introspected;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.Builder;
+
+import java.time.Instant;
+import java.util.List;
+
+@Singleton
+public class PluginUsageReport extends AbstractReportable<PluginUsageReport.PluginUsageEvent> {
+    
+    private final PluginRegistry pluginRegistry;
+    private final boolean enabled;
+    @Inject
+    public PluginUsageReport(PluginRegistry pluginRegistry) {
+        super(Types.PLUGIN_USAGE, Schedules.daily(), false);
+        this.pluginRegistry = pluginRegistry;
+        
+        ServerType serverType = KestraContext.getContext().getServerType();
+        this.enabled = ServerType.EXECUTOR.equals(serverType) || ServerType.STANDALONE.equals(serverType);
+    }
+    
+    @Override
+    public PluginUsageEvent report(final Instant now, final TimeInterval period) {
+        return PluginUsageEvent
+            .builder()
+            .plugins(PluginUsage.of(pluginRegistry))
+            .build();
+    }
+    
+    @Override
+    public boolean isEnabled() {
+        return enabled;
+    }
+    
+    @Builder
+    @Introspected
+    public record PluginUsageEvent(
+        List<PluginUsage> plugins
+    ) implements Event {
+    }
+}

core/src/main/java/io/kestra/core/reporter/reports/ServiceUsageReport.java

@@ -0,0 +1,53 @@
+package io.kestra.core.reporter.reports;
+
+import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.models.ServerType;
+import io.kestra.core.models.collectors.ServiceUsage;
+import io.kestra.core.reporter.AbstractReportable;
+import io.kestra.core.reporter.Schedules;
+import io.kestra.core.reporter.Types;
+import io.kestra.core.repositories.ServiceInstanceRepositoryInterface;
+import io.micronaut.core.annotation.Introspected;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.Builder;
+
+import java.time.Duration;
+import java.time.Instant;
+
+@Singleton
+public class ServiceUsageReport extends AbstractReportable<ServiceUsageReport.ServiceUsageEvent> {
+    
+    private final ServiceInstanceRepositoryInterface serviceInstanceRepository;
+    private final boolean isEnabled;
+    
+    @Inject
+    public ServiceUsageReport(ServiceInstanceRepositoryInterface serviceInstanceRepository) {
+        super(Types.SERVICE_USAGE, Schedules.daily(), false);
+        this.serviceInstanceRepository = serviceInstanceRepository;
+        
+        ServerType serverType = KestraContext.getContext().getServerType();
+        this.isEnabled = ServerType.STANDALONE.equals(serverType) || ServerType.EXECUTOR.equals(serverType);
+    }
+    
+    @Override
+    public ServiceUsageEvent report(final Instant now, final TimeInterval period) {
+        
+        return ServiceUsageEvent
+            .builder()
+            .services(ServiceUsage.of(period.from().toInstant(), period.to().toInstant(), serviceInstanceRepository, Duration.ofMinutes(5)))
+            .build();
+    }
+    
+    @Override
+    public boolean isEnabled() {
+        return isEnabled;
+    }
+    
+    @Builder
+    @Introspected
+    public record ServiceUsageEvent(
+        ServiceUsage services
+    ) implements Event {
+    }
+}

core/src/main/java/io/kestra/core/reporter/reports/SystemInformationReport.java

@@ -0,0 +1,63 @@
+package io.kestra.core.reporter.reports;
+
+import io.kestra.core.models.collectors.ConfigurationUsage;
+import io.kestra.core.models.collectors.HostUsage;
+import io.kestra.core.reporter.AbstractReportable;
+import io.kestra.core.reporter.Schedules;
+import io.kestra.core.reporter.Types;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.env.Environment;
+import io.micronaut.core.annotation.Introspected;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.Builder;
+
+import java.lang.management.ManagementFactory;
+import java.time.Instant;
+import java.util.Set;
+
+@Singleton
+public class SystemInformationReport extends AbstractReportable<SystemInformationReport.SystemInformationEvent> {
+    
+    private final Environment environment;
+    private final ApplicationContext applicationContext;
+    private final String kestraUrl;
+    private final Instant startTime;
+    
+    @Inject
+    public SystemInformationReport(ApplicationContext applicationContext) {
+        super(Types.SYSTEM_INFORMATION, Schedules.daily(), false);
+        this.environment = applicationContext.getEnvironment();
+        this.applicationContext = applicationContext;
+        this.kestraUrl = applicationContext.getProperty("kestra.url", String.class).orElse(null);
+        this.startTime = Instant.ofEpochMilli(ManagementFactory.getRuntimeMXBean().getStartTime());
+    }
+    
+    @Override
+    public SystemInformationEvent report(final Instant now, final TimeInterval timeInterval) {
+        return SystemInformationEvent
+            .builder()
+            .environments(environment.getActiveNames())
+            .configurations(ConfigurationUsage.of(applicationContext))
+            .startTime(startTime)
+            .host(HostUsage.of())
+            .uri(kestraUrl)
+            .build();
+    }
+    
+    @Override
+    public boolean isEnabled() {
+        return true;
+    }
+    
+    @Builder
+    @Introspected
+    public record SystemInformationEvent(
+        Set<String> environments,
+        HostUsage host,
+        ConfigurationUsage configurations,
+        Instant startTime,
+        String uri
+    ) implements Event {
+    }
+}

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

@@ -16,6 +16,14 @@ import java.util.Map;
 import java.util.Optional;
 
 public interface DashboardRepositoryInterface {
+    
+    /**
+     * Gets the total number of Dashboards.
+     *
+     * @return the total number.
+     */
+    long count();
+    
     Boolean isEnabled();
 
     Optional<Dashboard> get(String tenantId, String id);

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/runners/Executor.java

@@ -11,6 +11,10 @@ import lombok.Getter;
 import java.util.ArrayList;
 import java.util.List;
 
+// TODO for 2.0: this class is used as a queue consumer (which should have been the ExecutorInterface instead),
+//  a queue message (only in Kafka) and an execution context.
+//  At some point, we should rename it to ExecutorContext and move it to the executor module,
+//  then rename the ExecutorInterface to just Executor (to be used as a queue consumer)
 @Getter
 @AllArgsConstructor
 public class Executor {

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

@@ -1,7 +1,7 @@
 package io.kestra.core.runners;
 
-import java.io.Closeable;
+import io.kestra.core.server.Service;
 
-public interface ExecutorInterface extends Closeable, Runnable {
+public interface ExecutorInterface extends Service, Runnable {
 
 }

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

@@ -2,7 +2,6 @@ package io.kestra.core.runners;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableMap;
 import io.kestra.core.encryption.EncryptionService;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.KestraRuntimeException;
@@ -12,11 +11,14 @@ import io.kestra.core.models.flows.DependsOn;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.Input;
+import io.kestra.core.models.flows.Output;
 import io.kestra.core.models.flows.RenderableInput;
 import io.kestra.core.models.flows.Type;
 import io.kestra.core.models.flows.input.FileInput;
 import io.kestra.core.models.flows.input.InputAndValue;
 import io.kestra.core.models.flows.input.ItemTypeInterface;
+import io.kestra.core.models.property.Property;
+import io.kestra.core.models.property.PropertyContext;
 import io.kestra.core.models.property.URIFetcher;
 import io.kestra.core.models.tasks.common.EncryptedString;
 import io.kestra.core.models.validations.ManualConstraintViolation;
@@ -75,16 +77,19 @@ public class FlowInputOutput {
     private final StorageInterface storageInterface;
     private final Optional<String> secretKey;
     private final RunContextFactory runContextFactory;
+    private final VariableRenderer variableRenderer;
 
     @Inject
     public FlowInputOutput(
         StorageInterface storageInterface,
         RunContextFactory runContextFactory,
+        VariableRenderer variableRenderer,
         @Nullable @Value("${kestra.encryption.secret-key}") String secretKey
     ) {
         this.storageInterface = storageInterface;
         this.runContextFactory = runContextFactory;
         this.secretKey = Optional.ofNullable(secretKey);
+        this.variableRenderer = variableRenderer;
     }
 
     /**
@@ -249,11 +254,7 @@ public class FlowInputOutput {
         }
 
         final Map<String, ResolvableInput> resolvableInputMap = Collections.unmodifiableMap(inputs.stream()
-            .map(input -> {
-                // get value or default
-                Object value = Optional.ofNullable((Object) data.get(input.getId())).orElseGet(input::getDefaults);
-                return ResolvableInput.of(input, value);
-            })
+            .map(input -> ResolvableInput.of(input,data.get(input.getId())))
             .collect(Collectors.toMap(it -> it.get().input().getId(), Function.identity(), (o1, o2) -> o1, LinkedHashMap::new)));
 
         resolvableInputMap.values().forEach(input -> resolveInputValue(input, flow, execution, resolvableInputMap));
@@ -312,8 +313,16 @@ 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
-            final Object value = resolvable.get().value();
             if (value == null) {
                 if (input.getRequired()) {
                     resolvable.resolveWithError(input.toConstraintViolationException("missing required input", null));
@@ -341,12 +350,47 @@ 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);
+            case INT -> resolveDefaultPropertyAs(input, renderer, Integer.class);
+            case FLOAT -> resolveDefaultPropertyAs(input, renderer, Float.class);
+            case BOOLEAN, BOOL -> resolveDefaultPropertyAs(input, renderer, Boolean.class);
+            case DATETIME -> resolveDefaultPropertyAs(input, renderer, Instant.class);
+            case DATE -> resolveDefaultPropertyAs(input, renderer, LocalDate.class);
+            case TIME -> resolveDefaultPropertyAs(input, renderer, LocalTime.class);
+            case DURATION -> resolveDefaultPropertyAs(input, renderer, Duration.class);
+            case FILE, URI -> resolveDefaultPropertyAs(input, renderer, URI.class);
+            case JSON, YAML -> resolveDefaultPropertyAs(input, renderer, Object.class);
+            case ARRAY -> resolveDefaultPropertyAsList(input, renderer, Object.class);
+            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);
+    }
+    @SuppressWarnings("unchecked")
+    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()
             .collect(HashMap::new, (m, v) -> m.put(v.getKey(), v.getValue().value()), HashMap::putAll)
         );
+        // Hack: Pre-inject all inputs that have a default value with 'null' to prevent
+        // RunContextFactory from attempting to render them when absent, which could
+        // otherwise cause an exception if a Pebble expression is involved.
+        List<Input<?>> inputs = Optional.ofNullable(flow).map(FlowInterface::getInputs).orElse(List.of());
+        for (Input<?> input : inputs) {
+            if (input.getDefaults() != null && !flattenInputs.containsKey(input.getId())) {
+                flattenInputs.put(input.getId(), null);
+            }
+        }
         return runContextFactory.of(flow, execution, vars -> vars.withInputs(flattenInputs));
     }
 
@@ -368,7 +412,7 @@ public class FlowInputOutput {
         final Map<String, Object> in
     ) {
         if (flow.getOutputs() == null) {
-            return ImmutableMap.of();
+            return Map.of();
         }
         Map<String, Object> results = flow
             .getOutputs()
@@ -376,6 +420,9 @@ public class FlowInputOutput {
             .map(output -> {
                 Object current = in == null ? null : in.get(output.getId());
                 try {
+                    if (current == null && Boolean.FALSE.equals(output.getRequired())) {
+                        return Optional.of(new AbstractMap.SimpleEntry<>(output.getId(), null));
+                    }
                     return parseData(execution, output, current)
                         .map(entry -> {
                             if (output.getType().equals(Type.SECRET)) {
@@ -406,7 +453,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<>(
@@ -448,7 +495,7 @@ public class FlowInputOutput {
                 case URI -> {
                     Matcher matcher = URI_PATTERN.matcher(current.toString());
                     if (matcher.matches()) {
-                        yield current;
+                        yield current.toString();
                     } else {
                         throw new IllegalArgumentException("Expected `URI` but received `" + current + "`");
                     }
@@ -483,6 +530,30 @@ 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())) {
+                try {
+                    outputsById.putAll(runContext.render(Map.of(output.getId(), output.getValue())));
+                } catch (Exception e) {
+                    runContext.logger().warn("Failed to render optional flow output '{}'. Output is ignored.", output.getId(), e);
+                    outputsById.put(output.getId(), null);
+                }
+            }
+        }
+        return outputsById;
+    }
 
     /**
      * Mutable wrapper to hold a flow's input, and it's resolved value.
@@ -511,22 +582,26 @@ public class FlowInputOutput {
             return input;
         }
 
+        public void isDefault(boolean isDefault) {
+            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.exception());
+            this.input = new InputAndValue(input, this.input.value(), this.input.enabled(), this.input.isDefault(), this.input.exception());
         }
 
         public void resolveWithEnabled(boolean enabled) {
-            this.input = new InputAndValue(this.input.input(), input.value(), enabled, this.input.exception());
+            this.input = new InputAndValue(this.input.input(), input.value(), enabled, this.input.isDefault(), this.input.exception());
             markAsResolved();
         }
 
         public void resolveWithValue(@Nullable Object value) {
-            this.input = new InputAndValue(this.input.input(), value,  this.input.enabled(),  this.input.exception());
+            this.input = new InputAndValue(this.input.input(), value,  this.input.enabled(), this.input.isDefault(), this.input.exception());
             markAsResolved();
         }
 
         public void resolveWithError(@Nullable ConstraintViolationException exception) {
-            this.input = new InputAndValue(this.input.input(),  this.input.value(),  this.input.enabled(), exception);
+            this.input = new InputAndValue(this.input.input(),  this.input.value(), this.input.enabled(), this.input.isDefault(), exception);
             markAsResolved();
         }
 

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

@@ -49,6 +49,19 @@ public class FlowableUtils {
         return FlowableUtils.innerResolveSequentialNexts(execution, currentTasks, parentTaskRun);
     }
 
+    public static List<NextTaskRun> resolveSequentialNexts(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        List<ResolvedTask> _finally,
+        TaskRun parentTaskRun,
+        State.Type terminalState
+    ) {
+        List<ResolvedTask> currentTasks = execution.findTaskDependingFlowState(tasks, errors, _finally, parentTaskRun, terminalState);
+
+        return FlowableUtils.innerResolveSequentialNexts(execution, currentTasks, parentTaskRun);
+    }
+
     private static List<NextTaskRun> innerResolveSequentialNexts(
         Execution execution,
         List<ResolvedTask> currentTasks,
@@ -149,7 +162,31 @@ public class FlowableUtils {
         boolean allowFailure,
         boolean allowWarning
     ) {
-        List<ResolvedTask> currentTasks = execution.findTaskDependingFlowState(tasks, errors, _finally, parentTaskRun);
+        return resolveState(
+            execution,
+            tasks,
+            errors,
+            _finally,
+            parentTaskRun,
+            runContext,
+            allowFailure,
+            allowWarning,
+            State.Type.SUCCESS
+        );
+    }
+
+    public static Optional<State.Type> resolveState(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        List<ResolvedTask> _finally,
+        TaskRun parentTaskRun,
+        RunContext runContext,
+        boolean allowFailure,
+        boolean allowWarning,
+        State.Type terminalState
+    ) {
+        List<ResolvedTask> currentTasks = execution.findTaskDependingFlowState(tasks, errors, _finally, parentTaskRun, terminalState);
 
         if (currentTasks == null) {
             runContext.logger().warn(
@@ -161,17 +198,17 @@ public class FlowableUtils {
 
             return Optional.of(allowFailure ? allowWarning ? State.Type.SUCCESS : State.Type.WARNING : State.Type.FAILED);
         } else if (currentTasks.stream().allMatch(t -> t.getTask().getDisabled()) && !currentTasks.isEmpty()) {
-            // if all child tasks are disabled, we end in SUCCESS
-            return Optional.of(State.Type.SUCCESS);
+            // if all child tasks are disabled, we end in the terminal state
+            return Optional.of(terminalState);
         } else if (!currentTasks.isEmpty()) {
-            // handle nominal case, tasks or errors flow are ready to be analysed
+            // handle nominal case, tasks or errors flow are ready to be analyzed
             if (execution.isTerminated(currentTasks, parentTaskRun)) {
-                return Optional.of(execution.guessFinalState(tasks, parentTaskRun, allowFailure, allowWarning));
+                return Optional.of(execution.guessFinalState(tasks, parentTaskRun, allowFailure, allowWarning, terminalState));
             }
         } else {
             // first call, the error flow is not ready, we need to notify the parent task that can be failed to init error flows
-            if (execution.hasFailed(tasks, parentTaskRun)) {
-                return Optional.of(execution.guessFinalState(tasks, parentTaskRun, allowFailure, allowWarning));
+            if (execution.hasFailed(tasks, parentTaskRun) || terminalState == State.Type.FAILED) {
+                return Optional.of(execution.guessFinalState(tasks, parentTaskRun, allowFailure, allowWarning, terminalState));
             }
         }
 

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

@@ -1,4 +1,7 @@
 package io.kestra.core.runners;
-// NOTE: this class is not used anymore but must be kept as it is used in as queue consumer both in JDBC and Kafka
-public class Indexer {
+
+import io.kestra.core.server.Service;
+
+public interface Indexer extends Service, Runnable {
+
 }

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

@@ -6,6 +6,7 @@ import io.kestra.core.encryption.EncryptionService;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.executions.AbstractMetricEntry;
 import io.kestra.core.models.property.Property;
+import io.kestra.core.models.property.PropertyContext;
 import io.kestra.core.storages.StateStore;
 import io.kestra.core.storages.Storage;
 import io.kestra.core.storages.kv.KVStore;
@@ -18,7 +19,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
-public abstract class RunContext {
+public abstract class RunContext implements PropertyContext {
 
     /**
      * Returns the trigger execution id attached to this context.

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

@@ -9,6 +9,7 @@ import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.Type;
+import io.kestra.core.models.property.PropertyContext;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.plugins.PluginConfigurations;
@@ -77,7 +78,7 @@ public class RunContextFactory {
     public RunContextInitializer initializer() {
         return applicationContext.getBean(RunContextInitializer.class);
     }
-
+    
     public RunContext of(FlowInterface flow, Execution execution) {
         return of(flow, execution, Function.identity());
     }
@@ -98,7 +99,7 @@ public class RunContextFactory {
                     .withDecryptVariables(true)
                     .withSecretInputs(secretInputsFromFlow(flow))
                 )
-                .build(runContextLogger))
+                .build(runContextLogger, PropertyContext.create(variableRenderer)))
             .withSecretInputs(secretInputsFromFlow(flow))
             .build();
     }
@@ -127,7 +128,7 @@ public class RunContextFactory {
                 .withTaskRun(taskRun)
                 .withDecryptVariables(decryptVariables)
                 .withSecretInputs(secretInputsFromFlow(flow))
-                .build(runContextLogger))
+                .build(runContextLogger, PropertyContext.create(variableRenderer)))
             .withKvStoreService(kvStoreService)
             .withSecretInputs(secretInputsFromFlow(flow))
             .withTask(task)
@@ -146,7 +147,7 @@ public class RunContextFactory {
                 .withFlow(flow)
                 .withTrigger(trigger)
                 .withSecretInputs(secretInputsFromFlow(flow))
-                .build(runContextLogger)
+                .build(runContextLogger, PropertyContext.create(variableRenderer))
             )
             .withSecretInputs(secretInputsFromFlow(flow))
             .withTrigger(trigger)

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

@@ -20,16 +20,18 @@ import io.kestra.core.queues.QueueInterface;
 import jakarta.annotation.Nullable;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.IOUtils;
 import org.slf4j.LoggerFactory;
 
 import java.io.*;
+import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.*;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 public class RunContextLogger implements Supplier<org.slf4j.Logger> {
-    private static final int MAX_MESSAGE_LENGTH = 1024 * 10;
+    private static final int MAX_MESSAGE_LENGTH = 1024 * 15;
     public static final String ORIGINAL_TIMESTAMP_KEY = "originalTimestamp";
 
     private final String loggerName;
@@ -80,7 +82,6 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
         }
 
         List<LogEntry> result = new ArrayList<>();
-        long i = 0;
         for (String s : split) {
             result.add(LogEntry.builder()
                 .namespace(logEntry.getNamespace())
@@ -98,7 +99,6 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
                 .thread(event.getThreadName())
                 .build()
             );
-            i++;
         }
 
         return result;
@@ -144,8 +144,9 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
     }
 
     public void usedSecret(String secret) {
-        if (secret != null) {
+        if (secret != null && !secret.isEmpty()) {
             this.useSecrets.add(secret);
+            this.useSecrets.add(Base64.getEncoder().encodeToString(secret.getBytes(StandardCharsets.UTF_8)));
         }
     }
 
@@ -331,14 +332,11 @@ public class RunContextLogger implements Supplier<org.slf4j.Logger> {
         protected void append(ILoggingEvent e) {
             e = this.transform(e);
 
-            logEntries(e, logEntry)
-                .forEach(l -> {
-                    try {
-                        logQueue.emitAsync(l);
-                    } catch (QueueException ex) {
-                        log.warn("Unable to emit logQueue", ex);
-                    }
-                });
+            try {
+                logQueue.emitAsync(logEntries(e, logEntry));
+            } catch (QueueException ex) {
+                log.warn("Unable to emit logQueue", ex);
+            }
         }
     }
 

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

@@ -4,15 +4,11 @@ import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
-import jakarta.validation.ConstraintViolation;
-import jakarta.validation.ConstraintViolationException;
-import jakarta.validation.Validator;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 
 import static io.kestra.core.utils.Rethrow.throwFunction;
 
@@ -27,12 +23,19 @@ public class RunContextProperty<T> {
     private final RunContext runContext;
     private final Task task;
     private final AbstractTrigger trigger;
+    
+    private final boolean skipCache;
 
     RunContextProperty(Property<T> property, RunContext runContext) {
+        this(property, runContext, false);
+    }
+    
+    RunContextProperty(Property<T> property, RunContext runContext, boolean skipCache) {
         this.property = property;
         this.runContext = runContext;
         this.task = ((DefaultRunContext) runContext).getTask();
         this.trigger = ((DefaultRunContext) runContext).getTrigger();
+        this.skipCache = skipCache;
     }
 
     private void validate() {
@@ -45,6 +48,19 @@ public class RunContextProperty<T> {
             log.trace("Unable to do validation: no task or trigger found");
         }
     }
+    
+    /**
+     * Returns a new {@link RunContextProperty} that will always be rendered by evaluating
+     * its original Pebble expression, without using any previously cached value.
+     * <p>
+     * This ensures that each time the property is rendered, the underlying
+     * expression is re-evaluated to produce a fresh result.
+     *
+     * @return a new {@link Property} that bypasses the cache
+     */
+    public RunContextProperty<T> skipCache() {
+        return new RunContextProperty<>(this.property, this.runContext, true);
+    }
 
     /**
      * Render a property then convert it to its target type and validate it.<br>
@@ -55,13 +71,13 @@ public class RunContextProperty<T> {
      * Warning, due to the caching mechanism, this method is not thread-safe.
      */
     public Optional<T> as(Class<T> clazz) throws IllegalVariableEvaluationException {
-        var as = Optional.ofNullable(this.property)
+        var as = Optional.ofNullable(getProperty())
             .map(throwFunction(prop -> Property.as(prop, this.runContext, clazz)));
 
         validate();
         return as;
     }
-
+    
     /**
      * Render a property with additional variables, then convert it to its target type and validate it.<br>
      *
@@ -71,7 +87,7 @@ public class RunContextProperty<T> {
      * Warning, due to the caching mechanism, this method is not thread-safe.
      */
     public Optional<T> as(Class<T> clazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
-        var as = Optional.ofNullable(this.property)
+        var as = Optional.ofNullable(getProperty())
             .map(throwFunction(prop -> Property.as(prop, this.runContext, clazz, variables)));
 
         validate();
@@ -89,7 +105,7 @@ public class RunContextProperty<T> {
      */
     @SuppressWarnings("unchecked")
     public <I> T asList(Class<I> itemClazz) throws IllegalVariableEvaluationException {
-        var as = Optional.ofNullable(this.property)
+        var as = Optional.ofNullable(getProperty())
             .map(throwFunction(prop -> Property.asList(prop, this.runContext, itemClazz)))
             .orElse((T) Collections.emptyList());
 
@@ -108,7 +124,7 @@ public class RunContextProperty<T> {
      */
     @SuppressWarnings("unchecked")
     public <I> T asList(Class<I> itemClazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
-        var as = Optional.ofNullable(this.property)
+        var as = Optional.ofNullable(getProperty())
             .map(throwFunction(prop -> Property.asList(prop, this.runContext, itemClazz, variables)))
             .orElse((T) Collections.emptyList());
 
@@ -127,7 +143,7 @@ public class RunContextProperty<T> {
      */
     @SuppressWarnings("unchecked")
     public <K,V> T asMap(Class<K> keyClass, Class<V> valueClass) throws IllegalVariableEvaluationException {
-        var as = Optional.ofNullable(this.property)
+        var as = Optional.ofNullable(getProperty())
             .map(throwFunction(prop -> Property.asMap(prop, this.runContext, keyClass, valueClass)))
             .orElse((T) Collections.emptyMap());
 
@@ -146,11 +162,15 @@ public class RunContextProperty<T> {
      */
     @SuppressWarnings("unchecked")
     public <K,V> T asMap(Class<K> keyClass, Class<V> valueClass, Map<String, Object> variables) throws IllegalVariableEvaluationException {
-        var as = Optional.ofNullable(this.property)
+        var as = Optional.ofNullable(getProperty())
             .map(throwFunction(prop -> Property.asMap(prop, this.runContext, keyClass, valueClass, variables)))
             .orElse((T) Collections.emptyMap());
 
         validate();
         return as;
     }
+    
+    private Property<T> getProperty() {
+        return skipCache ? this.property.skipCache() : this.property;
+    }
 }

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

@@ -1,6 +1,7 @@
 package io.kestra.core.runners;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
@@ -9,6 +10,8 @@ import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.flows.input.SecretInput;
+import io.kestra.core.models.property.Property;
+import io.kestra.core.models.property.PropertyContext;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.utils.ListUtils;
@@ -138,10 +141,10 @@ public final class RunVariables {
          * @param logger    The {@link RunContextLogger logger}
          * @return          The immutable map of variables.
          */
-        Map<String, Object> build(final RunContextLogger logger);
+        Map<String, Object> build(RunContextLogger logger, PropertyContext propertyContext);
     }
 
-    public record  KestraConfiguration(String environment, String url) { }
+    public record KestraConfiguration(String environment, String url) { }
 
     /**
      * Default builder class for constructing variables.
@@ -174,7 +177,7 @@ public final class RunVariables {
 
         // Note: for performance reason, cloning maps should be avoided as much as possible.
         @Override
-        public Map<String, Object> build(final RunContextLogger logger) {
+        public Map<String, Object> build(final RunContextLogger logger, final PropertyContext propertyContext) {
             ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
 
             builder.put("envs", envs != null ? envs : Map.of());
@@ -282,7 +285,13 @@ public final class RunVariables {
                     // we add default inputs value from the flow if not already set, this will be useful for triggers
                     flow.getInputs().stream()
                         .filter(input -> input.getDefaults() != null && !inputs.containsKey(input.getId()))
-                        .forEach(input -> inputs.put(input.getId(), input.getDefaults()));
+                        .forEach(input -> {
+                            try {
+                                inputs.put(input.getId(), FlowInputOutput.resolveDefaultValue(input, propertyContext));
+                            } catch (IllegalVariableEvaluationException e) {
+                                // Silent catch, if an input depends on another input, or a variable that is populated at runtime / input filling time, we can't resolve it here.
+                            }
+                        });
                 }
 
                 if (!inputs.isEmpty()) {

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

@@ -1,5 +0,0 @@
-package io.kestra.core.runners;
-
-public interface RunnerInterface {
-    void run();
-}

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

@@ -1,4 +1,4 @@
-package io.kestra.core.schedulers;
+package io.kestra.core.runners;
 
 import java.util.function.Consumer;
 

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

@@ -2,6 +2,5 @@ package io.kestra.core.runners;
 
 import io.kestra.core.server.Service;
 
-public interface IndexerInterface extends Service, Runnable {
-
+public interface Scheduler extends Service, Runnable {
 }

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

@@ -1,4 +1,4 @@
-package io.kestra.core.schedulers;
+package io.kestra.core.runners;
 
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.flows.Flow;