feat(jdbc): Improve internal queue cleaning

Instead of cleaning queues via the JdbcCleaner, or via queues.deleteByIds(), directly clean some queues after processing.
We only do this for queues that are known to have a single consumer, for these queues, instead of updating the offsets after consumption, we remove directly the records.

.devcontainer/Dockerfile

@@ -0,0 +1,67 @@
+FROM ubuntu:24.04
+
+ARG DEBIAN_FRONTEND=noninteractive
+
+USER root
+WORKDIR /root
+
+RUN apt update && apt install -y \
+  apt-transport-https ca-certificates gnupg curl wget git zip unzip less zsh net-tools iputils-ping jq lsof
+
+ENV HOME="/root"
+
+# --------------------------------------
+# Git
+# --------------------------------------
+# Need to add the devcontainer workspace folder as a safe directory to enable git 
+# version control system to be enabled in the containers file system.
+RUN git config --global --add safe.directory "/workspaces/kestra"
+# --------------------------------------
+
+# --------------------------------------
+# Oh my zsh
+# --------------------------------------
+RUN sh -c "$(curl -fsSL https://raw.githubusercontent.com/ohmyzsh/ohmyzsh/master/tools/install.sh)" -- \
+  -t robbyrussell \
+  -p git -p node -p npm
+
+ENV SHELL=/bin/zsh
+# --------------------------------------
+
+# --------------------------------------
+# Java
+# --------------------------------------
+RUN wget https://download.oracle.com/java/21/latest/jdk-21_linux-x64_bin.deb
+RUN dpkg -i ./jdk-21_linux-x64_bin.deb
+ENV JAVA_HOME=/usr/java/jdk-21-oracle-x64
+ENV PATH="$PATH:$JAVA_HOME/bin"
+# Will load a custom configuration file for Micronaut
+ENV MICRONAUT_ENVIRONMENTS=local,override
+# Sets the path where you save plugins as Jar and is loaded during the startup process
+ENV KESTRA_PLUGINS_PATH="/workspaces/kestra/local/plugins"
+# --------------------------------------
+
+# --------------------------------------
+# Node.js
+# --------------------------------------
+RUN curl -fsSL https://deb.nodesource.com/setup_22.x -o nodesource_setup.sh \
+  && bash nodesource_setup.sh && apt install -y nodejs
+# Increases JavaScript heap memory to 4GB to prevent heap out of error during startup
+ENV NODE_OPTIONS=--max-old-space-size=4096
+# --------------------------------------
+
+# --------------------------------------
+# Python
+# --------------------------------------
+RUN apt install -y python3 pip python3-venv
+# --------------------------------------
+
+# --------------------------------------
+# SSH 
+# --------------------------------------
+RUN mkdir -p ~/.ssh
+RUN touch ~/.ssh/config
+RUN echo "Host github.com" >> ~/.ssh/config \
+  && echo "    IdentityFile ~/.ssh/id_ed25519" >> ~/.ssh/config
+RUN touch ~/.ssh/id_ed25519
+# --------------------------------------

.devcontainer/README.md

@@ -0,0 +1,149 @@
+# Kestra Devcontainer
+
+This devcontainer provides a quick and easy setup for anyone using VSCode to get up and running quickly with this project to start development on either the frontend or backend. It bootstraps a docker container for you to develop inside of without the need to manually setup the environment.
+
+---
+
+## INSTRUCTIONS
+
+### Setup:
+
+Take a look at this guide to get an idea of what the setup is like as this devcontainer setup follows this approach: https://kestra.io/docs/getting-started/contributing
+
+Once you have this repo cloned to your local system, you will need to install the VSCode extension [Remote Development](https://marketplace.visualstudio.com/items?itemName=ms-vscode-remote.vscode-remote-extensionpack).
+
+Then run the following command from the command palette:
+`Dev Containers: Open Folder in Container...` and select your Kestra root folder.
+
+This will then put you inside a docker container ready for development.
+
+NOTE: you'll need to wait for the gradle build to finish and compile Java files but this process should happen automatically within VSCode.
+
+In the meantime, you can move onto the next step...
+
+---
+
+### Development:
+
+- Create a `.env.development.local` file in the `ui` folder and paste the following:
+
+```bash
+# This lets the frontend know what the backend URL is but you are free to change this to your actual server URL e.g. hosted version of Kestra.
+VITE_APP_API_URL=http://localhost:8080
+```
+
+- Navigate into the `ui` folder and run `npm install` to install the dependencies for the frontend project.
+
+- Now go to the `cli/src/main/resources` folder and create a `application-override.yml` file.
+
+Now you have two choices:
+
+`Local mode`:
+
+Runs the Kestra server in local mode which uses a H2 database, so this is the only config you'd need:
+
+```yaml
+micronaut:
+  server:
+    cors:
+      enabled: true
+      configurations:
+        all:
+          allowedOrigins:
+            - http://localhost:5173
+```
+
+You can then open a new terminal and run the following command to start the backend server: `./gradlew runLocal`
+
+`Standalone mode`:
+
+Runs in standalone mode which uses Postgres. Make sure to have a local Postgres instance already running on localhost:
+
+```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
+  server:
+    basic-auth:
+      enabled: false
+
+datasources:
+  postgres:
+    # It is important to note that you must use the "host.docker.internal" host when connecting to a docker container outside of your devcontainer as attempting to use localhost will only point back to this devcontainer.
+    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
+      # We must ignore missing migrations as we may delete the wrong ones or delete those that are not used anymore.
+      ignore-migration-patterns: "*:missing,*:future"
+      out-of-order: true
+
+micronaut:
+  server:
+    cors:
+      enabled: true
+      configurations:
+        all:
+          allowedOrigins:
+            - http://localhost:5173
+```
+
+Then add the following settings to the `.vscode/launch.json` file:
+
+```json
+{
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "type": "java",
+      "name": "Kestra Standalone",
+      "request": "launch",
+      "mainClass": "io.kestra.cli.App",
+      "projectName": "cli",
+      "args": "server standalone"
+    }
+  ]
+}
+```
+
+You can then use the VSCode `Run and Debug` extension to start the Kestra server.
+
+Additionally, if you're doing frontend development, you can run `npm run dev` from the `ui` folder after having the above running (which will provide a backend) to access your application from `localhost:5173`. This has the benefit to watch your changes and hot-reload upon doing frontend changes.
+
+#### Plugins
+If you want your plugins to be loaded inside your devcontainer, point the `source` field to a folder containing jars of the plugins you want to embed in the following snippet in `devcontainer.json`:
+```
+"mounts": [
+  {
+    "source": "/absolute/path/to/your/local/jar/plugins/folder",
+    "target": "/workspaces/kestra/local/plugins",
+    "type": "bind"
+  }
+],
+```
+
+---
+
+### GIT
+
+If you want to commit to GitHub, make sure to navigate to the `~/.ssh` folder and either create a new SSH key or override the existing `id_ed25519` file and paste an existing SSH key from your local machine into this file. You will then need to change the permissions of the file by running: `chmod 600 id_ed25519`. This will allow you to then push to GitHub.
+
+---

.devcontainer/devcontainer.json

@@ -0,0 +1,46 @@
+{
+  "name": "kestra",
+  "build": {
+    "context": ".",
+    "dockerfile": "Dockerfile"
+  },
+  "workspaceFolder": "/workspaces/kestra",
+  "forwardPorts": [5173, 8080],
+  "customizations": {
+    "vscode": {
+      "settings": {
+        "terminal.integrated.profiles.linux": {
+          "zsh": {
+            "path": "/bin/zsh"
+          }
+        },
+        "workbench.iconTheme": "vscode-icons",
+        "editor.tabSize": 4,
+        "editor.formatOnSave": true,
+        "files.insertFinalNewline": true,
+        "editor.defaultFormatter": "esbenp.prettier-vscode",
+        "telemetry.telemetryLevel": "off",
+        "editor.bracketPairColorization.enabled": true,
+        "editor.guides.bracketPairs": "active"
+      },
+      "extensions": [
+        "redhat.vscode-yaml",
+        "dbaeumer.vscode-eslint",
+        "vscode-icons-team.vscode-icons",
+        "eamodio.gitlens",
+        "esbenp.prettier-vscode",
+        "aaron-bond.better-comments",
+        "codeandstuff.package-json-upgrade",
+        "andys8.jest-snippets",
+        "oderwat.indent-rainbow",
+        "evondev.indent-rainbow-palettes",
+        "formulahendry.auto-rename-tag",
+        "IronGeek.vscode-env",
+        "yoavbls.pretty-ts-errors",
+        "github.vscode-github-actions",
+        "vscjava.vscode-java-pack",
+        "ms-azuretools.vscode-docker"
+      ]
+    }
+  }
+}

.github/actions/generate-translations/action.yml

@@ -1,57 +0,0 @@
-name: Generate Translations
-description: "Required the environment variable OPENAI_API_KEY to be set. This action will generate translations for the UI."
-
-inputs:
-    github-token:
-        description: 'GitHub Token'
-        required: true
-
-runs:
-  using: composite
-
-  steps:
-    - name: Checkout code
-      uses: actions/checkout@v4
-      with:
-        ref: ${{ github.head_ref }}
-
-    - name: Set up Python
-      uses: actions/setup-python@v5
-      with:
-        python-version: "3.x"
-
-    - name: Install Python dependencies
-      shell: bash
-      run: pip install gitpython openai
-
-    - name: Generate translations
-      shell: bash
-      run: python ui/src/translations/generate_translations.py
-
-    - name: Set up Node
-      uses: actions/setup-node@v4
-      with:
-        node-version: "20.x"
-
-    - name: Check keys matching
-      shell: bash
-      run: node ui/src/translations/check.js
-
-    - name: Set up Git
-      shell: bash
-      run: |
-        git config --global user.name "GitHub Action"
-        git config --global user.email "actions@github.com"
-
-    - name: Check for changes and commit
-      env:
-        GH_TOKEN: ${{ inputs.github-token }}
-      shell: bash
-      run: |
-        git add ui/src/translations/*.json
-        if git diff --cached --quiet; then
-          echo "No changes to commit. Exiting with success."
-          exit 0
-        fi
-        git commit -m "chore(translations): auto generate values for languages other than english"
-        git push origin ${{ github.head_ref }}

.github/workflows/docker.yml

@@ -47,11 +47,11 @@ jobs:
         image:
           - name: "-no-plugins"
             plugins: ""
-            packages: ""
+            packages: jattach
             python-libs: ""
           - name: ""
             plugins: ${{needs.plugins.outputs.plugins}}
-            packages: python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip
+            packages: python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach
             python-libs: kestra
     steps:
       - uses: actions/checkout@v4
@@ -63,11 +63,11 @@ jobs:
           if [[ "${{ inputs.release-tag }}" == "" ]]; then
             TAG=${GITHUB_REF#refs/*/}
             echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          else 
+          else
             TAG="${{ inputs.release-tag }}"
             echo "tag=${TAG}" >> $GITHUB_OUTPUT
-          fi 
-         
+          fi
+
           if [[ "${{ env.PLUGIN_VERSION }}" == *"-SNAPSHOT" ]]; then
             echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots ${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT;
           else
@@ -75,7 +75,7 @@ jobs:
           fi
       # Download release
       - name: Download release
-        uses: robinraju/release-downloader@v1.11
+        uses: robinraju/release-downloader@v1.12
         with:
           tag: ${{steps.vars.outputs.tag}}
           fileName: 'kestra-*'
@@ -89,6 +89,11 @@ jobs:
       - 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
 

.github/workflows/generate-translations.yml

@@ -0,0 +1,67 @@
+name: Auto-Translate UI keys and create PR
+
+on:
+  schedule:
+    - cron: "0 9-21 * * *"  # Every hour from 9 AM to 9 PM
+  workflow_dispatch:
+    inputs:
+      retranslate_modified_keys:
+        description: "Whether to re-translate modified keys even if they already have translations."
+        type: choice
+        options:
+          - "false"
+          - "true"
+        default: "false"
+        required: false
+
+jobs:
+  translations:
+    name: Translations
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+        - uses: actions/checkout@v4
+          name: Checkout
+          with:
+            fetch-depth: 0
+
+        - name: Set up Python
+          uses: actions/setup-python@v5
+          with:
+            python-version: "3.x"
+
+        - name: Install Python dependencies
+          run: pip install gitpython openai
+
+        - name: Generate translations
+          run: python ui/src/translations/generate_translations.py ${{ github.event.inputs.retranslate_modified_keys }}
+          env:
+            OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+
+        - name: Set up Node
+          uses: actions/setup-node@v4
+          with:
+            node-version: "20.x"
+
+        - name: Check keys matching
+          run: node ui/src/translations/check.js
+
+        - name: Set up Git
+          run: |
+            git config --global user.name "GitHub Action"
+            git config --global user.email "actions@github.com"
+
+        - name: Commit and create PR
+          env:
+            GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          run: |
+            BRANCH_NAME="chore/update-translations-$(date +%s)"
+            git checkout -b $BRANCH_NAME
+            git add ui/src/translations/*.json
+            if git diff --cached --quiet; then
+              echo "No changes to commit. Exiting with success."
+              exit 0
+            fi
+            git commit -m "chore(translations): localize to languages other than English"
+            git push -u origin $BRANCH_NAME || (git push origin --delete $BRANCH_NAME && git push -u origin $BRANCH_NAME)
+            gh pr create --title "Translations from en.json" --body "This PR was created automatically by a GitHub Action." --base develop --head $BRANCH_NAME --assignee anna-geller --reviewer anna-geller

.github/workflows/main.yml

@@ -18,7 +18,7 @@ on:
       - v*
 
 concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
+  group: ${{ github.workflow }}-${{ github.ref }}-main
   cancel-in-progress: true
 
 jobs:
@@ -31,6 +31,7 @@ jobs:
   release:
     name: Release
     needs: [tests]
+    if: "!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' }}

.github/workflows/pull-request.yml

@@ -6,11 +6,15 @@ on:
       - develop
 
 concurrency:
-  group: ${{ github.workflow }}-${{ github.ref_name }}
+  group: ${{ github.workflow }}-${{ github.ref_name }}-pr
   cancel-in-progress: true
 
 jobs:
+  # ********************************************************************************************************************
+  # File changes detection
+  # ********************************************************************************************************************
   file-changes:
+    if: ${{ github.event.pull_request.draft == false }}
     name: File changes detection
     runs-on: ubuntu-latest
     timeout-minutes: 60
@@ -25,34 +29,16 @@ jobs:
           filters: |
             ui:
               - 'ui/**'
-            translations:
-              - 'ui/src/translations/**'
             backend:
               - '!{ui,.github}/**'
           token: ${{ secrets.GITHUB_TOKEN }}
 
-  translations:
-    name: 'Translations - Generate translations'
-    runs-on: ubuntu-latest
-    timeout-minutes: 60
-    needs: file-changes
-    steps:
-        - uses: actions/checkout@v4
-          name: Checkout - Current ref
-          if: "needs.file-changes.outputs.translations == 'true'"
-
-        - id: generate-translations
-          name: Translations - Generate translations
-          if: "needs.file-changes.outputs.translations == 'true'"
-          uses: ./.github/actions/generate-translations
-          with:
-            github-token: ${{ secrets.GITHUB_TOKEN }}
-          env:
-            OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
-
+  # ********************************************************************************************************************
+  # Tests
+  # ********************************************************************************************************************
   frontend:
     name: Frontend - Tests
-    needs: [file-changes, translations]
+    needs: [file-changes]
     if: "needs.file-changes.outputs.ui == 'true'"
     uses: ./.github/workflows/workflow-frontend-test.yml
     secrets:

.github/workflows/setversion-tag.yml

@@ -23,12 +23,11 @@ jobs:
             exit 1
           fi
           
-          CURRENT_BRANCH="{{ github.ref }}"
-          
           # Extract the major and minor versions
           BASE_VERSION=$(echo "$RELEASE_VERSION" | sed -E 's/^([0-9]+\.[0-9]+)\..*/\1/')
           RELEASE_BRANCH="refs/heads/releases/v${BASE_VERSION}.x"
           
+          CURRENT_BRANCH="$GITHUB_REF"
           if ! [[ "$CURRENT_BRANCH" == "$RELEASE_BRANCH" ]]; then
             echo "Invalid release branch. Expected $RELEASE_BRANCH, was $CURRENT_BRANCH"
             exit 1

.github/workflows/vulnerabilities-check.yml

@@ -37,7 +37,7 @@ jobs:
       - name: Npm - Install
         shell: bash
         working-directory: ui
-        run: npm install
+        run: npm ci
 
       # Run OWASP dependency check plugin
       - name: Gradle Dependency Check
@@ -80,7 +80,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.29.0
+        uses: aquasecurity/trivy-action@0.30.0
         with:
           image-ref: kestra/kestra:develop
           format: table
@@ -113,7 +113,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.29.0
+        uses: aquasecurity/trivy-action@0.30.0
         with:
           image-ref: kestra/kestra:latest
           format: table

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

@@ -16,6 +16,11 @@ on:
         description: 'Google Service Account'
         required: true
 
+permissions:
+  contents: write
+  checks: write
+  actions: read
+
 jobs:
   test:
     name: Backend - Tests
@@ -54,7 +59,7 @@ jobs:
 
       # report test
       - name: Test - Publish Test Results
-        uses: dorny/test-reporter@v1
+        uses: dorny/test-reporter@v2
         if: always()
         with:
           name: Java Tests Report
@@ -63,6 +68,7 @@ jobs:
           list-suites: 'failed'
           list-tests: 'failed'
           fail-on-error: 'false'
+          token: ${{ secrets.GITHUB_AUTH_TOKEN }}
 
       # Sonar
       - name: Test - Analyze with Sonar

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

@@ -107,6 +107,11 @@ jobs:
       - name: Docker - Setup QEMU
         uses: docker/setup-qemu-action@v3
 
+      - name: Docker - Fix Qemu
+        shell: bash
+        run: |
+          docker run --rm --privileged multiarch/qemu-user-static --reset -p yes -c yes
+
       - name: Docker - Setup Buildx
         uses: docker/setup-buildx-action@v3
 

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

@@ -20,17 +20,23 @@ jobs:
           name: exe
           path: build/executable
 
-      # GitHub Release
-      - name: GitHub - Create release
-        id: create_github_release
-        uses: "marvinpinto/action-automatic-releases@latest"
-        if: startsWith(github.ref, 'refs/tags/v')
-        continue-on-error: true
+      # Checkout GitHub Actions
+      - name: Checkout - Actions
+        uses: actions/checkout@v4
         with:
-          repo_token: "${{ secrets.GITHUB_TOKEN }}"
-          prerelease: false
-          files: |
-            build/executable/*
+          repository: kestra-io/actions
+          sparse-checkout-cone-mode: true
+          path: actions
+          sparse-checkout: |
+            .github/actions
+
+      # GitHub Release
+      - name: Create GitHub release
+        uses: ./actions/.github/actions/github-release
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
 
       # Trigger gha workflow to bump helm chart version
       - name: GitHub - Trigger the Helm chart version bump

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

@@ -1,7 +1,30 @@
 name: Publish - Docker
 
 on:
+  workflow_dispatch:
+    inputs:
+      plugin-version:
+        description: "Kestra version"
+        default: 'LATEST'
+        required: false
+        type: string
+      force-download-artifact:
+        description: 'Force download artifact'
+        required: false
+        type: string
+        default: "true"
   workflow_call:
+    inputs:
+      plugin-version:
+        description: "Kestra version"
+        default: 'LATEST'
+        required: false
+        type: string
+      force-download-artifact:
+        description: 'Force download artifact'
+        required: false
+        type: string
+        default: "true"
     secrets:
       DOCKERHUB_USERNAME:
         description: "The Dockerhub username."
@@ -11,26 +34,39 @@ on:
         required: true
 
 jobs:
+  # ********************************************************************************************************************
+  # Build
+  # ********************************************************************************************************************
   build-artifacts:
-    name: Build - Artifacts
+    name: Build Artifacts
+    if: ${{ github.event.inputs.force-download-artifact == 'true' }}
     uses: ./.github/workflows/workflow-build-artifacts.yml
     with:
       plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
-
+  # ********************************************************************************************************************
+  # Docker
+  # ********************************************************************************************************************
   publish:
     name: Publish - Docker
-    needs: build-artifacts
     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: ${{ needs.build-artifacts.outputs.docker-tag }}-no-plugins
-            packages: ""
+          - tag: -no-plugins
+            packages: jattach
+            plugins: false
             python-libraries: ""
 
-          - tag: ${{ needs.build-artifacts.outputs.docker-tag }}
-            plugins: ${{ needs.build-artifacts.outputs.plugins }}
-            packages: python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip
+          - tag: ""
+            plugins: true
+            packages: python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach
             python-libraries: kestra
     steps:
       - name: Checkout - Current ref
@@ -40,6 +76,11 @@ jobs:
       - name: Docker - Setup QEMU
         uses: docker/setup-qemu-action@v3
 
+      - name: Docker - Fix Qemu
+        shell: bash
+        run: |
+          docker run --rm --privileged multiarch/qemu-user-static --reset -p yes -c yes
+
       - name: Docker - Setup Docker Buildx
         uses: docker/setup-buildx-action@v3
 
@@ -50,17 +91,34 @@ jobs:
           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 image name
+      - name: Docker - Set variables
         shell: bash
         id: vars
         run: |
           TAG=${GITHUB_REF#refs/*/}
-          if [[ $TAG = "master" || $TAG == v* ]]; then
+          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://s01.oss.sonatype.org/content/repositories/snapshots $PLUGINS" >> $GITHUB_OUTPUT
           else
-            echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots ${{ matrix.image.plugins }}" >> $GITHUB_OUTPUT
+            TAG="build-${{ github.run_id }}";
+            echo "plugins=--repositories=https://s01.oss.sonatype.org/content/repositories/snapshots $PLUGINS" >> $GITHUB_OUTPUT
           fi
+          
+          echo "tag=${TAG}${{ matrix.image.tag }}" >> $GITHUB_OUTPUT
 
       # Build Docker Image
       - name: Artifacts - Download executable
@@ -80,7 +138,7 @@ jobs:
         with:
           context: .
           push: true
-          tags: kestra/kestra:${{ matrix.image.tag }}
+          tags: kestra/kestra:${{ steps.vars.outputs.tag }}
           platforms: linux/amd64,linux/arm64
           build-args: |
             KESTRA_PLUGINS=${{ steps.vars.outputs.plugins }}

.github/workflows/workflow-release.yml

@@ -1,6 +1,18 @@
 name: Release
 
 on:
+  workflow_dispatch:
+    inputs:
+      plugin-version:
+        description: "Kestra version"
+        default: 'LATEST'
+        required: false
+        type: string
+      publish-docker:
+        description: "Publish Docker image"
+        default: 'false'
+        required: false
+        type: string
   workflow_call:
     inputs:
       plugin-version:
@@ -31,12 +43,23 @@ on:
         description: "The Sonatype GPG file."
         required: true
 jobs:
+  build-artifacts:
+    name: Build - Artifacts
+    uses: ./.github/workflows/workflow-build-artifacts.yml
+    with:
+      plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
+
   Docker:
     name: Publish Docker
+    needs: build-artifacts
     uses: ./.github/workflows/workflow-publish-docker.yml
+    if: startsWith(github.ref, 'refs/heads/develop') || github.event.inputs.publish-docker == 'true'
+    with:
+      force-download-artifact: 'false'
+      plugin-version: ${{ github.event.inputs.plugin-version != null && github.event.inputs.plugin-version || 'LATEST' }}
     secrets:
-        DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
-        DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
+      DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
+      DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
 
   Maven:
     name: Publish Maven
@@ -50,6 +73,7 @@ jobs:
 
   Github:
     name: Github Release
+    needs: build-artifacts
     if: startsWith(github.ref, 'refs/tags/v')
     uses: ./.github/workflows/workflow-github-release.yml
     secrets:

.github/workflows/workflow-test.yml

@@ -19,8 +19,31 @@ on:
         value: ${{ jobs.set-backend-status.outputs.backend_status }}
 
 jobs:
+  file-changes:
+    name: File changes detection
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    outputs:
+      ui: ${{ steps.changes.outputs.ui }}
+      backend: ${{ steps.changes.outputs.backend }}
+    steps:
+      - uses: actions/checkout@v4
+        if: "!startsWith(github.ref, 'refs/tags/v')"
+      - uses: dorny/paths-filter@v3
+        if: "!startsWith(github.ref, 'refs/tags/v')"
+        id: changes
+        with:
+          filters: |
+            ui:
+              - 'ui/**'
+            backend:
+              - '!{ui,.github}/**'
+          token: ${{ secrets.GITHUB_TOKEN }}
+
   frontend:
     name: Frontend - Tests
+    needs: file-changes
+    if: "needs.file-changes.outputs.ui == 'true' || startsWith(github.ref, 'refs/tags/v')"
     uses: ./.github/workflows/workflow-frontend-test.yml
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
@@ -29,6 +52,8 @@ jobs:
 
   backend:
     name: Backend - Tests
+    needs: file-changes
+    if: "needs.file-changes.outputs.backend == 'true' || startsWith(github.ref, 'refs/tags/v')"
     uses: ./.github/workflows/workflow-backend-test.yml
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -37,6 +37,7 @@ ui/coverage
 ui/stats.html
 ui/.frontend-gradle-plugin
 ui/utils/CHANGELOG.md
+ui/test-report.junit.xml
 
 ### Docker
 /.env
@@ -57,3 +58,4 @@ core/src/main/resources/gradle.properties
 **/allure-results/*
 
 *storybook.log
+storybook-static

.plugins

@@ -32,6 +32,7 @@
 #plugin-git:io.kestra.plugin:plugin-git:LATEST
 #plugin-github:io.kestra.plugin:plugin-github:LATEST
 #plugin-googleworkspace:io.kestra.plugin:plugin-googleworkspace:LATEST
+#plugin-graalvm:io.kestra.plugin:plugin-graalvm:LATEST
 #plugin-hightouch:io.kestra.plugin:plugin-hightouch:LATEST
 #plugin-hubspot:io.kestra.plugin:plugin-hubspot:LATEST
 #plugin-huggingface:io.kestra.plugin:plugin-huggingface:LATEST

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" \
+		--build-arg="APT_PACKAGES=python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach" \
 		--build-arg="PYTHON_LIBRARIES=kestra" \
 		-t ${DOCKER_IMAGE}:${VERSION} ${DOCKER_PATH} || exit 1 ;
 

README.md

@@ -24,6 +24,13 @@
 <a href="https://www.youtube.com/@kestra-io"><img height="25" src="https://kestra.io/youtube.svg" alt="youtube" /></a> &nbsp;
 </p>
 
+<p align="center">
+  <a href="https://trendshift.io/repositories/2714" target="_blank">
+    <img src="https://trendshift.io/api/badge/repositories/2714" alt="kestra-io%2Fkestra | Trendshift" width="250" height="55"/>
+  </a>
+  <a href="https://www.producthunt.com/posts/kestra?embed=true&utm_source=badge-top-post-badge&utm_medium=badge&utm_souce=badge-kestra" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/top-post-badge.svg?post_id=612077&theme=light&period=daily&t=1740737506162" alt="Kestra - All&#0045;in&#0045;one&#0032;automation&#0032;&#0038;&#0032;orchestration&#0032;platform | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
+</p>
+
 <p align="center">
     <a href="https://go.kestra.io/video/product-overview" target="_blank">
         <img src="https://kestra.io/startvideo.png" alt="Get started in 4 minutes with Kestra" width="640px" />
@@ -47,7 +54,7 @@ Kestra is an open-source, event-driven orchestration platform that makes both **
 - **Structure & Resilience**: tame chaos and bring resilience to your workflows with **namespaces**, **labels**, **subflows**, **retries**, **timeout**, **error handling**, **inputs**, **outputs** that generate artifacts in the UI, **variables**, **conditional branching**, **advanced scheduling**, **event triggers**, **backfills**, **dynamic tasks**, **sequential and parallel tasks**, and skip tasks or triggers when needed by setting the flag `disabled` to `true`.
 
 
-🧑‍💻 The YAML definition gets automatically adjusted any time you make changes to a workflow from the UI or via an API call. Therefore, the orchestration logic is **always managed declaratively in code**, even if you modify your workflows in other ways (UI, CI/CD, Terraform, API calls). 
+🧑‍💻 The YAML definition gets automatically adjusted any time you make changes to a workflow from the UI or via an API call. Therefore, the orchestration logic is **always managed declaratively in code**, even if you modify your workflows in other ways (UI, CI/CD, Terraform, API calls).
 
 
 <p align="center">

build.gradle

@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "6.0.1.5171"
+    id "org.sonarqube" version "6.1.0.5360"
     id 'jacoco-report-aggregation'
 
     // helper
@@ -33,13 +33,13 @@ plugins {
     // release
     id "io.github.gradle-nexus.publish-plugin" version "2.0.0"
     id 'net.researchgate.release' version '3.1.0'
-    id "com.gorylenko.gradle-git-properties" version "2.4.2"
+    id "com.gorylenko.gradle-git-properties" version "2.5.0"
     id 'signing'
     id 'ru.vyarus.pom' version '3.0.0' apply false
     id 'ru.vyarus.github-info' version '2.0.0' apply false
 
     // OWASP dependency check
-    id "org.owasp.dependencycheck" version "12.0.2" apply false
+    id "org.owasp.dependencycheck" version "12.1.1" apply false
 }
 
 idea {
@@ -74,7 +74,7 @@ dependencies {
  **********************************************************************************************************************/
 allprojects {
     if (it.name != 'platform') {
-        group "io.kestra"
+        group = "io.kestra"
 
         java {
             sourceCompatibility = targetJavaVersion
@@ -121,7 +121,6 @@ allprojects {
             micronaut "io.micronaut:micronaut-management"
             micronaut "io.micrometer:micrometer-core"
             micronaut "io.micronaut.micrometer:micronaut-micrometer-registry-prometheus"
-            micronaut "io.micronaut.micrometer:micronaut-micrometer-registry-otlp"
             micronaut "io.micronaut:micronaut-http-client"
             micronaut "io.micronaut.reactor:micronaut-reactor-http-client"
             micronaut "io.micronaut.tracing:micronaut-tracing-opentelemetry-http"
@@ -280,7 +279,7 @@ subprojects {
         }
 
         dependencies {
-            agent "org.aspectj:aspectjweaver:1.9.22.1"
+            agent "org.aspectj:aspectjweaver:1.9.23"
         }
 
         test {
@@ -597,11 +596,10 @@ release {
     }
 
     // Dynamically set properties with default values
-    failOnSnapshotDependencies = (project.hasProperty('release.failOnSnapshotDependencies')
-            ? project.property('release.failOnSnapshotDependencies').toBoolean()
-            : true)
+    failOnSnapshotDependencies = providers.gradleProperty("release.failOnSnapshotDependencies")
+            .map(val -> Boolean.parseBoolean(val))
+            .getOrElse(true)
 
-    pushReleaseVersionBranch = (project.hasProperty('release.pushReleaseVersionBranch')
-            ? project.property('release.pushReleaseVersionBranch').toString()
-            : null)
+    pushReleaseVersionBranch = providers.gradleProperty("release.pushReleaseVersionBranch")
+            .getOrElse(null)
 }

cli/build.gradle

@@ -12,18 +12,9 @@ dependencies {
     implementation 'ch.qos.logback.contrib:logback-json-classic'
     implementation 'ch.qos.logback.contrib:logback-jackson'
 
-    // plugins
-    implementation 'org.eclipse.aether:aether-api'
-    implementation 'org.eclipse.aether:aether-spi'
-    implementation 'org.eclipse.aether:aether-util'
-    implementation 'org.eclipse.aether:aether-impl'
-    implementation 'org.eclipse.aether:aether-connector-basic'
-    implementation 'org.eclipse.aether:aether-transport-file'
-    implementation 'org.eclipse.aether:aether-transport-http'
-    implementation('org.apache.maven:maven-aether-provider') {
-        // sisu dependency injector is not used
-        exclude group: 'org.eclipse.sisu'
-    }
+    // OTLP metrics
+    implementation "io.micronaut.micrometer:micronaut-micrometer-registry-otlp"
+
     // aether still use javax.inject
     compileOnly 'javax.inject:javax.inject:1'
 
@@ -43,4 +34,7 @@ dependencies {
     implementation project(":storage-local")
 
     implementation project(":webserver")
+
+    //test
+    testImplementation "org.wiremock:wiremock"
 }

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

@@ -46,8 +46,18 @@ public abstract class AbstractApiCommand extends AbstractCommand {
     @Nullable
     private HttpClientConfiguration httpClientConfiguration;
 
+    /**
+     * {@inheritDoc}
+     */
+    protected boolean loadExternalPlugins() {
+        return false;
+    }
+
     protected DefaultHttpClient client() throws URISyntaxException {
-        DefaultHttpClient defaultHttpClient = new DefaultHttpClient(server.toURI(), httpClientConfiguration != null ? httpClientConfiguration : new DefaultHttpClientConfiguration());
+        DefaultHttpClient defaultHttpClient = DefaultHttpClient.builder()
+            .uri(server.toURI())
+            .configuration(httpClientConfiguration != null ? httpClientConfiguration : new DefaultHttpClientConfiguration())
+            .build();
         MessageBodyHandlerRegistry defaultHandlerRegistry = defaultHttpClient.getHandlerRegistry();
         if (defaultHandlerRegistry instanceof ContextlessMessageBodyHandlerRegistry modifiableRegistry) {
             modifiableRegistry.add(MediaType.TEXT_JSON_TYPE, new NettyJsonHandler<>(JsonMapper.createDefault()));

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

@@ -4,16 +4,17 @@ import ch.qos.logback.classic.LoggerContext;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.cli.commands.servers.ServerCommandInterface;
 import io.kestra.cli.services.StartupHookInterface;
-import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.plugins.PluginManager;
 import io.kestra.core.plugins.PluginRegistry;
 import io.kestra.webserver.services.FlowAutoLoaderService;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.env.yaml.YamlPropertySourceLoader;
 import io.micronaut.core.annotation.Introspected;
+import io.micronaut.http.uri.UriBuilder;
 import io.micronaut.management.endpoint.EndpointDefaultConfiguration;
 import io.micronaut.runtime.server.EmbeddedServer;
+import jakarta.inject.Provider;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.http.client.utils.URIBuilder;
 import io.kestra.core.utils.Rethrow;
 import picocli.CommandLine;
 
@@ -26,10 +27,13 @@ import java.nio.file.Paths;
 import java.text.MessageFormat;
 import java.time.temporal.ChronoUnit;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import jakarta.inject.Inject;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
 
-@CommandLine.Command(
+@Command(
     versionProvider = VersionProvider.class,
     mixinStandardHelpOptions = true,
     showDefaultValues = true
@@ -49,22 +53,28 @@ abstract public class AbstractCommand implements Callable<Integer> {
     @Inject
     private io.kestra.core.utils.VersionProvider versionProvider;
 
+    @Inject
+    protected Provider<PluginRegistry> pluginRegistryProvider;
+
+    @Inject
+    protected Provider<PluginManager> pluginManagerProvider;
+
     private PluginRegistry pluginRegistry;
 
-    @CommandLine.Option(names = {"-v", "--verbose"}, description = "Change log level. Multiple -v options increase the verbosity.", showDefaultValue = CommandLine.Help.Visibility.NEVER)
+    @Option(names = {"-v", "--verbose"}, description = "Change log level. Multiple -v options increase the verbosity.", showDefaultValue = CommandLine.Help.Visibility.NEVER)
     private boolean[] verbose = new boolean[0];
 
-    @CommandLine.Option(names = {"-l", "--log-level"}, description = "Change log level (values: ${COMPLETION-CANDIDATES})")
+    @Option(names = {"-l", "--log-level"}, description = "Change log level (values: ${COMPLETION-CANDIDATES})")
     private LogLevel logLevel = LogLevel.INFO;
 
-    @CommandLine.Option(names = {"--internal-log"}, description = "Change also log level for internal log")
+    @Option(names = {"--internal-log"}, description = "Change also log level for internal log")
     private boolean internalLog = false;
 
-    @CommandLine.Option(names = {"-c", "--config"}, description = "Path to a configuration file")
+    @Option(names = {"-c", "--config"}, description = "Path to a configuration file")
     private Path config = Paths.get(System.getProperty("user.home"), ".kestra/config.yml");
 
-    @CommandLine.Option(names = {"-p", "--plugins"}, description = "Path to plugins directory")
-    protected Path pluginsPath = System.getenv("KESTRA_PLUGINS_PATH") != null ? Paths.get(System.getenv("KESTRA_PLUGINS_PATH")) : null;
+    @Option(names = {"-p", "--plugins"}, description = "Path to plugins directory")
+    protected Path pluginsPath = Optional.ofNullable(System.getenv("KESTRA_PLUGINS_PATH")).map(Paths::get).orElse(null);
 
     public enum LogLevel {
         TRACE,
@@ -76,7 +86,7 @@ abstract public class AbstractCommand implements Callable<Integer> {
 
     @Override
     public Integer call() throws Exception {
-        Thread.currentThread().setName(this.getClass().getDeclaredAnnotation(CommandLine.Command.class).name());
+        Thread.currentThread().setName(this.getClass().getDeclaredAnnotation(Command.class).name());
         startLogger();
         sendServerLog();
         if (this.startupHook != null) {
@@ -84,8 +94,14 @@ abstract public class AbstractCommand implements Callable<Integer> {
         }
 
         if (this.pluginsPath != null && loadExternalPlugins()) {
-            pluginRegistry = pluginRegistry();
+            pluginRegistry = pluginRegistryProvider.get();
             pluginRegistry.registerIfAbsent(pluginsPath);
+
+            // PluginManager mus only be initialized if a registry is also instantiated
+            if (isPluginManagerEnabled()) {
+                PluginManager manager = pluginManagerProvider.get();
+                manager.start();
+            }
         }
 
         startWebserver();
@@ -102,8 +118,15 @@ abstract public class AbstractCommand implements Callable<Integer> {
         return true;
     }
 
-    protected PluginRegistry pluginRegistry() {
-        return KestraContext.getContext().getPluginRegistry(); // Lazy init
+    /**
+     * Specifies whether the {@link PluginManager} service must be initialized.
+     * <p>
+     * This method can be overridden by concrete commands.
+     *
+     * @return {@code true} if the {@link PluginManager} service must be initialized.
+     */
+    protected boolean isPluginManagerEnabled() {
+        return true;
     }
 
     private static String message(String message, Object... format) {
@@ -157,7 +180,6 @@ abstract public class AbstractCommand implements Callable<Integer> {
                         logger.getName().startsWith("io.kestra") &&
                             !logger.getName().startsWith("io.kestra.ee.runner.kafka.services"))
                 )
-                    || logger.getName().startsWith("flow")
             )
             .forEach(
                 logger -> logger.setLevel(ch.qos.logback.classic.Level.valueOf(this.logLevel.name()))
@@ -183,9 +205,9 @@ abstract public class AbstractCommand implements Callable<Integer> {
                 if (this.endpointConfiguration.getPort().isPresent()) {
                     URI endpoint = null;
                     try {
-                        endpoint = new URIBuilder(server.getURL().toURI())
-                            .setPort(this.endpointConfiguration.getPort().get())
-                            .setPath("/health")
+                        endpoint = UriBuilder.of(server.getURL().toURI())
+                            .port(this.endpointConfiguration.getPort().get())
+                            .path("/health")
                             .build();
                     } catch (URISyntaxException e) {
                         e.printStackTrace();
@@ -207,10 +229,12 @@ abstract public class AbstractCommand implements Callable<Integer> {
         return false;
     }
 
-    protected void shutdownHook(Rethrow.RunnableChecked<Exception> run) {
+    protected void shutdownHook(boolean logShutdown, Rethrow.RunnableChecked<Exception> run) {
         Runtime.getRuntime().addShutdownHook(new Thread(
             () -> {
-                log.warn("Receiving shutdown ! Try to graceful exit");
+                if (logShutdown) {
+                    log.warn("Receiving shutdown ! Try to graceful exit");
+                }
                 try {
                     run.run();
                 } catch (Exception e) {

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

@@ -31,6 +31,12 @@ public abstract class AbstractValidateCommand extends AbstractApiCommand {
     @CommandLine.Parameters(index = "0", description = "the directory containing files to check")
     protected Path directory;
 
+    /** {@inheritDoc} **/
+    @Override
+    protected boolean loadExternalPlugins() {
+        return local;
+    }
+
     public static void handleException(ConstraintViolationException e, String resource) {
         stdErr("\t@|fg(red) Unable to parse {0} due to the following error(s):|@", resource);
         e.getConstraintViolations()
@@ -68,10 +74,9 @@ public abstract class AbstractValidateCommand extends AbstractApiCommand {
         }
     }
 
-    // bug in micronaut, we can't inject YamlFlowParser & ModelValidator, so we inject from implementation
+    // bug in micronaut, we can't inject ModelValidator, so we inject from implementation
     public Integer call(
         Class<?> cls,
-        YamlParser yamlParser,
         ModelValidator modelValidator,
         Function<Object, String> identity,
         Function<Object, List<String>> warningsFunction,
@@ -88,7 +93,7 @@ public abstract class AbstractValidateCommand extends AbstractApiCommand {
                     .filter(YamlParser::isValidExtension)
                     .forEach(path -> {
                         try {
-                            Object parse = yamlParser.parse(path.toFile(), cls);
+                            Object parse = YamlParser.parse(path.toFile(), cls);
                             modelValidator.validate(parse);
                             stdOut("@|green \u2713|@ - " + identity.apply(parse));
                             List<String> warnings = warningsFunction.apply(parse);

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

@@ -18,6 +18,8 @@ import picocli.CommandLine;
         FlowNamespaceCommand.class,
         FlowDotCommand.class,
         FlowExportCommand.class,
+        FlowUpdateCommand.class,
+        FlowUpdatesCommand.class
     }
 )
 @Slf4j

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

@@ -29,8 +29,7 @@ public class FlowDotCommand extends AbstractCommand {
     public Integer call() throws Exception {
         super.call();
 
-        YamlParser parser = applicationContext.getBean(YamlParser.class);
-        Flow flow = parser.parse(file.toFile(), Flow.class);
+        Flow flow = YamlParser.parse(file.toFile(), Flow.class);
 
         GraphCluster graph = GraphUtils.of(flow, null);
 

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

@@ -20,9 +20,6 @@ public class FlowExpandCommand extends AbstractCommand {
     @CommandLine.Parameters(index = "0", description = "The flow file to expand")
     private Path file;
 
-    @Inject
-    private YamlParser yamlParser;
-
     @Inject
     private ModelValidator modelValidator;
 
@@ -31,7 +28,7 @@ public class FlowExpandCommand extends AbstractCommand {
         super.call();
         stdErr("Warning, this functionality is deprecated and will be removed at some point.");
         String content = IncludeHelperExpander.expand(Files.readString(file), file.getParent());
-        Flow flow = yamlParser.parse(content, Flow.class);
+        Flow flow = YamlParser.parse(content, Flow.class);
         modelValidator.validate(flow);
         stdOut(content);
         return 0;

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

@@ -33,6 +33,9 @@ public class FlowUpdatesCommand extends AbstractApiCommand {
     @CommandLine.Option(names = {"--delete"}, negatable = true, description = "Whether missing should be deleted")
     public boolean delete = false;
 
+    @CommandLine.Option(names = {"--namespace"}, description = "The parent namespace of the flows, if not set, every namespace are allowed.")
+    public String namespace;
+
     @SuppressWarnings("deprecation")
     @Override
     public Integer call() throws Exception {
@@ -58,8 +61,12 @@ public class FlowUpdatesCommand extends AbstractApiCommand {
                 body = String.join("\n---\n", flows);
             }
             try(DefaultHttpClient client = client()) {
+                String namespaceQuery = "";
+                if (namespace != null) {
+                    namespaceQuery = "&namespace=" + namespace;
+                }
                 MutableHttpRequest<String> request = HttpRequest
-                    .POST(apiUri("/flows/bulk") + "?delete=" + delete, body).contentType(MediaType.APPLICATION_YAML);
+                    .POST(apiUri("/flows/bulk") + "?allowNamespaceChild=true&delete=" + delete + namespaceQuery, body).contentType(MediaType.APPLICATION_YAML);
 
                 List<UpdateResult> updated = client.toBlocking().retrieve(
                     this.requestOptions(request),
@@ -80,4 +87,9 @@ public class FlowUpdatesCommand extends AbstractApiCommand {
 
         return 0;
     }
+
+    @Override
+    protected boolean loadExternalPlugins() {
+        return false;
+    }
 }

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

@@ -1,9 +1,8 @@
 package io.kestra.cli.commands.flows;
 
 import io.kestra.cli.AbstractValidateCommand;
-import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.validations.ModelValidator;
-import io.kestra.core.serializers.YamlParser;
 import io.kestra.core.services.FlowService;
 import jakarta.inject.Inject;
 import picocli.CommandLine;
@@ -16,8 +15,6 @@ import java.util.List;
     description = "Validate a flow"
 )
 public class FlowValidateCommand extends AbstractValidateCommand {
-    @Inject
-    private YamlParser yamlParser;
 
     @Inject
     private ModelValidator modelValidator;
@@ -28,23 +25,22 @@ public class FlowValidateCommand extends AbstractValidateCommand {
     @Override
     public Integer call() throws Exception {
         return this.call(
-            Flow.class,
-            yamlParser,
+            FlowWithSource.class,
             modelValidator,
             (Object object) -> {
-                Flow flow = (Flow) object;
+                FlowWithSource flow = (FlowWithSource) object;
                 return flow.getNamespace() + " / " + flow.getId();
             },
             (Object object) -> {
-                Flow flow = (Flow) object;
+                FlowWithSource flow = (FlowWithSource) object;
                 List<String> warnings = new ArrayList<>();
                 warnings.addAll(flowService.deprecationPaths(flow).stream().map(deprecation -> deprecation + " is deprecated").toList());
-                warnings.addAll(flowService.warnings(flow));
+                warnings.addAll(flowService.warnings(flow, this.tenantId));
                 return warnings;
             },
             (Object object) -> {
-                Flow flow = (Flow) object;
-                return flowService.relocations(flow.generateSource()).stream().map(relocation -> relocation.from() + " is replaced by " + relocation.to()).toList();
+                FlowWithSource flow = (FlowWithSource) object;
+                return flowService.relocations(flow.sourceOrGenerateIfNull()).stream().map(relocation -> relocation.from() + " is replaced by " + relocation.to()).toList();
             }
         );
     }

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

@@ -10,7 +10,6 @@ import io.micronaut.http.MediaType;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
-import jakarta.inject.Inject;
 import jakarta.validation.ConstraintViolationException;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
@@ -27,8 +26,6 @@ import java.util.List;
 )
 @Slf4j
 public class FlowNamespaceUpdateCommand extends AbstractServiceNamespaceUpdateCommand {
-    @Inject
-    public YamlParser yamlParser;
 
     @CommandLine.Option(names = {"--override-namespaces"}, negatable = true, description = "Replace namespace of all flows by the one provided")
     public boolean override = false;

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

@@ -1,31 +1,37 @@
 package io.kestra.cli.commands.plugins;
 
-import io.micronaut.configuration.picocli.PicocliRunner;
-import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
 import io.kestra.cli.AbstractCommand;
 import io.kestra.cli.App;
-import picocli.CommandLine;
+import io.micronaut.configuration.picocli.PicocliRunner;
+import lombok.SneakyThrows;
+import picocli.CommandLine.Command;
 
-@CommandLine.Command(
+@Command(
     name = "plugins",
     description = "Manage plugins",
     mixinStandardHelpOptions = true,
     subcommands = {
         PluginInstallCommand.class,
+        PluginUninstallCommand.class,
         PluginListCommand.class,
-        PluginDocCommand.class
+        PluginDocCommand.class,
+        PluginSearchCommand.class
     }
 )
-@Slf4j
 public class PluginCommand extends AbstractCommand {
+
     @SneakyThrows
     @Override
     public Integer call() throws Exception {
         super.call();
 
-        PicocliRunner.call(App.class, "plugins",  "--help");
+        PicocliRunner.call(App.class, "plugins", "--help");
 
         return 0;
     }
+
+    @Override
+    protected boolean loadExternalPlugins() {
+        return false;
+    }
 }

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

@@ -3,6 +3,7 @@ package io.kestra.cli.commands.plugins;
 import com.google.common.io.Files;
 import io.kestra.cli.AbstractCommand;
 import io.kestra.core.docs.DocumentationGenerator;
+import io.kestra.core.plugins.PluginRegistry;
 import io.kestra.core.plugins.RegisteredPlugin;
 import io.kestra.core.serializers.JacksonMapper;
 import io.micronaut.context.ApplicationContext;
@@ -42,8 +43,10 @@ public class PluginDocCommand extends AbstractCommand {
         super.call();
         DocumentationGenerator documentationGenerator = applicationContext.getBean(DocumentationGenerator.class);
 
-        List<RegisteredPlugin> plugins = core ?  pluginRegistry().plugins() : pluginRegistry().externalPlugins();
+        PluginRegistry registry = pluginRegistryProvider.get();
+        List<RegisteredPlugin> plugins = core ? registry.plugins() : registry.externalPlugins();
         boolean hasFailures = false;
+
         for (RegisteredPlugin registeredPlugin : plugins) {
             try {
                 documentationGenerator
@@ -100,4 +103,10 @@ public class PluginDocCommand extends AbstractCommand {
 
         return hasFailures ? 1 : 0;
     }
+
+    /** {@inheritDoc} **/
+    @Override
+    protected boolean isPluginManagerEnabled() {
+        return false;
+    }
 }

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

@@ -1,98 +1,123 @@
 package io.kestra.cli.commands.plugins;
 
-import org.apache.commons.io.FilenameUtils;
+import io.kestra.core.contexts.MavenPluginRepositoryConfig;
+import io.kestra.core.plugins.LocalPluginManager;
+import io.kestra.core.plugins.MavenPluginDownloader;
+import io.kestra.core.plugins.PluginArtifact;
+import io.kestra.core.plugins.PluginCatalogService;
+import io.kestra.core.plugins.PluginManager;
+import io.micronaut.http.client.HttpClient;
+import io.micronaut.http.client.annotation.Client;
+import io.micronaut.http.uri.UriBuilder;
 import io.kestra.cli.AbstractCommand;
-import io.kestra.cli.plugins.PluginDownloader;
-import io.kestra.cli.plugins.RepositoryConfig;
 import io.kestra.core.utils.IdUtils;
-import org.apache.http.client.utils.URIBuilder;
+import jakarta.inject.Provider;
 import picocli.CommandLine;
 
 import java.net.URI;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Objects;
 
 import jakarta.inject.Inject;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Parameters;
+import picocli.CommandLine.Option;
+import picocli.CommandLine.Spec;
 
-import static io.kestra.core.utils.Rethrow.throwConsumer;
-
-@CommandLine.Command(
+@Command(
     name = "install",
     description = "Install plugins"
 )
 public class PluginInstallCommand extends AbstractCommand {
-    @CommandLine.Parameters(index = "0..*", description = "Plugins to install. Represented as Maven artifact coordinates.")
+
+    @Option(names = {"--locally"}, description = "Specifies if plugins must be installed locally. If set to false the installation depends on your Kestra configuration.")
+    boolean locally = true;
+
+    @Option(names = {"--all"}, description = "Install all available plugins")
+    boolean all = false;
+
+    @Parameters(index = "0..*", description = "Plugins to install. Represented as Maven artifact coordinates (i.e., <groupId>:<artifactId>:(<version>|LATEST)")
     List<String> dependencies = new ArrayList<>();
 
-    @CommandLine.Option(names = {"--repositories"}, description = "URL to additional Maven repositories")
+    @Option(names = {"--repositories"}, description = "URL to additional Maven repositories")
     private URI[] repositories;
 
-    @CommandLine.Spec
+    @Spec
     CommandLine.Model.CommandSpec spec;
 
     @Inject
-    private PluginDownloader pluginDownloader;
+    Provider<MavenPluginDownloader> mavenPluginRepositoryProvider;
+
+    @Inject
+    @Client("api") HttpClient httpClient;
 
     @Override
     public Integer call() throws Exception {
         super.call();
 
-        if (this.pluginsPath == null) {
+        if (this.locally && this.pluginsPath == null) {
             throw new CommandLine.ParameterException(this.spec.commandLine(), "Missing required options '--plugins' " +
                 "or environment variable 'KESTRA_PLUGINS_PATH"
             );
         }
 
-        if (!pluginsPath.toFile().exists()) {
-            if (!pluginsPath.toFile().mkdir()) {
-                throw new RuntimeException("Cannot create directory: " + pluginsPath.toFile().getAbsolutePath());
-            }
-        }
-
+        List<MavenPluginRepositoryConfig> repositoryConfigs = List.of();
         if (repositories != null) {
-            Arrays.stream(repositories)
-                .forEach(throwConsumer(s -> {
-                    URIBuilder uriBuilder = new URIBuilder(s);
-
-                    RepositoryConfig.RepositoryConfigBuilder builder = RepositoryConfig.builder()
+            repositoryConfigs = Arrays.stream(repositories)
+                .map(uri -> {
+                    MavenPluginRepositoryConfig.MavenPluginRepositoryConfigBuilder builder = MavenPluginRepositoryConfig
+                        .builder()
                         .id(IdUtils.create());
 
-                    if (uriBuilder.getUserInfo() != null) {
-                        int index = uriBuilder.getUserInfo().indexOf(":");
-
-                        builder.basicAuth(new RepositoryConfig.BasicAuth(
-                            uriBuilder.getUserInfo().substring(0, index),
-                            uriBuilder.getUserInfo().substring(index + 1)
+                    String userInfo = uri.getUserInfo();
+                    if (userInfo != null) {
+                        String[] userInfoParts = userInfo.split(":");
+                        builder = builder.basicAuth(new MavenPluginRepositoryConfig.BasicAuth(
+                            userInfoParts[0],
+                            userInfoParts[1]
                         ));
-
-                        uriBuilder.setUserInfo(null);
                     }
-
-                    builder.url(uriBuilder.build().toString());
-
-                    pluginDownloader.addRepository(builder.build());
-                }));
+                    builder.url(UriBuilder.of(uri).userInfo(null).build().toString());
+                    return builder.build();
+                }).toList();
         }
 
-        List<URL> resolveUrl = pluginDownloader.resolve(dependencies);
-        stdOut("Resolved Plugin(s) with {0}", resolveUrl);
+        if (all) {
+            PluginCatalogService service = new PluginCatalogService(httpClient, false, true);
+            dependencies = service.get().stream().map(Objects::toString).toList();
+        }
 
-        for (URL url: resolveUrl) {
-            Files.copy(
-                Paths.get(url.toURI()),
-                Paths.get(pluginsPath.toString(), FilenameUtils.getName(url.toString())),
-                StandardCopyOption.REPLACE_EXISTING
+        if (dependencies.isEmpty()) {
+            stdErr("Error: No plugin to install.");
+            return CommandLine.ExitCode.OK;
+        }
+
+        final List<PluginArtifact> pluginArtifacts;
+        try {
+           pluginArtifacts = dependencies.stream().map(PluginArtifact::fromCoordinates).toList();
+        } catch (IllegalArgumentException e) {
+            stdErr(e.getMessage());
+            return CommandLine.ExitCode.USAGE;
+        }
+
+        try (final PluginManager pluginManager = getPluginManager()) {
+            List<PluginArtifact> installed = pluginManager.install(
+                pluginArtifacts,
+                repositoryConfigs,
+                false,
+                pluginsPath
             );
+
+            List<URI> uris = installed.stream().map(PluginArtifact::uri).toList();
+            stdOut("Successfully installed plugins {0} into {1}", dependencies, uris);
+            return CommandLine.ExitCode.OK;
         }
+    }
 
-        stdOut("Successfully installed plugins {0} into {1}", dependencies, pluginsPath);
-
-        return 0;
+    private PluginManager getPluginManager() {
+        return locally ? new LocalPluginManager(mavenPluginRepositoryProvider.get()) : this.pluginManagerProvider.get();
     }
 
     @Override

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

@@ -1,22 +1,31 @@
 package io.kestra.cli.commands.plugins;
 
 import io.kestra.cli.AbstractCommand;
+import io.kestra.core.plugins.PluginRegistry;
 import io.kestra.core.plugins.RegisteredPlugin;
+import jakarta.inject.Inject;
+import jakarta.inject.Provider;
 import picocli.CommandLine;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
+import picocli.CommandLine.Spec;
 
 import java.util.List;
 
-@CommandLine.Command(
+@Command(
     name = "list",
     description = "List all plugins already installed"
 )
 public class PluginListCommand extends AbstractCommand {
-    @CommandLine.Spec
+    @Spec
     CommandLine.Model.CommandSpec spec;
 
-    @CommandLine.Option(names = {"--core"}, description = "Also write core tasks plugins")
+    @Option(names = {"--core"}, description = "Also write core tasks plugins")
     private boolean core = false;
 
+    @Inject
+    private PluginRegistry registry;
+
     @Override
     public Integer call() throws Exception {
         super.call();
@@ -27,7 +36,8 @@ public class PluginListCommand extends AbstractCommand {
             );
         }
 
-        List<RegisteredPlugin> plugins = core ?  pluginRegistry().plugins() :  pluginRegistry().externalPlugins();
+        List<RegisteredPlugin> plugins = core ? registry.plugins() : registry.externalPlugins();
+
         plugins.forEach(registeredPlugin -> stdOut(registeredPlugin.toString()));
 
         return 0;

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

@@ -0,0 +1,149 @@
+package io.kestra.cli.commands.plugins;
+
+import io.kestra.cli.AbstractCommand;
+import io.micronaut.core.type.Argument;
+import io.micronaut.http.HttpRequest;
+import io.micronaut.http.client.HttpClient;
+import io.micronaut.http.client.annotation.Client;
+import jakarta.inject.Inject;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Parameters;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Command(
+    name = "search",
+    description = "Search for available Kestra plugins"
+)
+public class PluginSearchCommand extends AbstractCommand {
+    @Inject
+    @Client("api")
+    private HttpClient httpClient;
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+    private static final char SPACE = ' ';
+
+    @Parameters(index = "0", description = "Search term (optional)", defaultValue = "")
+    private String searchTerm;
+
+    @Override
+    public Integer call() throws Exception {
+        super.call();
+
+        try {
+            JsonNode root = fetchPlugins();
+            List<PluginInfo> plugins = findPlugins(root);
+            printResults(plugins);
+            return 0;
+        } catch (Exception e) {
+            stdOut("Error processing plugins: {0}", e.getMessage());
+            return 1;
+        }
+    }
+
+    private JsonNode fetchPlugins() throws Exception {
+        String response = httpClient.toBlocking()
+            .retrieve(
+                HttpRequest.GET("/v1/plugins")
+                    .header("Accept", "application/json")
+            );
+        return MAPPER.readTree(response);
+    }
+
+    private List<PluginInfo> findPlugins(JsonNode root) {
+        String searchTermLower = searchTerm.toLowerCase();
+        List<PluginInfo> plugins = new ArrayList<>();
+
+        for (JsonNode plugin : root) {
+            if (matchesSearch(plugin, searchTermLower)) {
+                plugins.add(new PluginInfo(
+                    plugin.path("name").asText(),
+                    plugin.path("title").asText(),
+                    plugin.path("group").asText(),
+                    plugin.path("version").asText("")
+                ));
+            }
+        }
+
+        plugins.sort((p1, p2) -> p1.name.compareToIgnoreCase(p2.name));
+        return plugins;
+    }
+
+    private boolean matchesSearch(JsonNode plugin, String term) {
+        if (term.isEmpty()) {
+            return true;
+        }
+
+        return plugin.path("name").asText().toLowerCase().contains(term) ||
+            plugin.path("title").asText().toLowerCase().contains(term) ||
+            plugin.path("group").asText().toLowerCase().contains(term);
+    }
+
+    private void printResults(List<PluginInfo> plugins) {
+        if (plugins.isEmpty()) {
+            stdOut("No plugins found{0}",
+                searchTerm.isEmpty() ? "" : " matching '" + searchTerm + "'");
+            return;
+        }
+
+        stdOut("\nFound {0} plugins{1}",
+            plugins.size(),
+            searchTerm.isEmpty() ? "" : " matching '" + searchTerm + "'"
+        );
+
+        printPluginsTable(plugins);
+    }
+
+    private void printPluginsTable(List<PluginInfo> plugins) {
+        int maxName = 4, maxTitle = 5, maxGroup = 5;
+        for (PluginInfo plugin : plugins) {
+            maxName = Math.max(maxName, plugin.name.length());
+            maxTitle = Math.max(maxTitle, plugin.title.length());
+            maxGroup = Math.max(maxGroup, plugin.group.length());
+        }
+
+        StringBuilder namePad = new StringBuilder(maxName);
+        StringBuilder titlePad = new StringBuilder(maxTitle);
+        StringBuilder groupPad = new StringBuilder(maxGroup);
+
+        stdOut("");
+        printRow(namePad, titlePad, groupPad, "NAME", "TITLE", "GROUP", "VERSION",
+            maxName, maxTitle, maxGroup);
+
+        for (PluginInfo plugin : plugins) {
+            printRow(namePad, titlePad, groupPad, plugin.name, plugin.title, plugin.group, plugin.version,
+                maxName, maxTitle, maxGroup);
+        }
+        stdOut("");
+    }
+
+    private void printRow(StringBuilder namePad, StringBuilder titlePad, StringBuilder groupPad,
+                          String name, String title, String group, String version,
+                          int maxName, int maxTitle, int maxGroup) {
+        stdOut("{0}  {1}  {2}  {3}",
+            pad(namePad, name, maxName),
+            pad(titlePad, title, maxTitle),
+            pad(groupPad, group, maxGroup),
+            version
+        );
+    }
+
+    private String pad(StringBuilder sb, String str, int length) {
+        sb.setLength(0);
+        sb.append(str);
+        while (sb.length() < length) {
+            sb.append(SPACE);
+        }
+        return sb.toString();
+    }
+
+    private record PluginInfo(String name, String title, String group, String version) {}
+
+    @Override
+    protected boolean loadExternalPlugins() {
+        return false;
+    }
+}

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

@@ -0,0 +1,69 @@
+package io.kestra.cli.commands.plugins;
+
+import io.kestra.cli.AbstractCommand;
+import io.kestra.core.plugins.LocalPluginManager;
+import io.kestra.core.plugins.MavenPluginDownloader;
+import io.kestra.core.plugins.PluginArtifact;
+import io.kestra.core.plugins.PluginManager;
+import jakarta.inject.Inject;
+import jakarta.inject.Provider;
+import picocli.CommandLine;
+import picocli.CommandLine.Parameters;
+import picocli.CommandLine.Spec;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+
+@CommandLine.Command(
+    name = "uninstall",
+    description = "Uninstall plugins"
+)
+public class PluginUninstallCommand extends AbstractCommand {
+    @Parameters(index = "0..*", description = "The plugins to uninstall. Represented as Maven artifact coordinates (i.e., <groupId>:<artifactId>:(<version>|LATEST)")
+    List<String> dependencies = new ArrayList<>();
+
+    @Spec
+    CommandLine.Model.CommandSpec spec;
+
+    @Inject
+    Provider<MavenPluginDownloader> mavenPluginRepositoryProvider;
+
+    @Override
+    public Integer call() throws Exception {
+        super.call();
+
+        List<PluginArtifact> pluginArtifacts;
+        try {
+            pluginArtifacts = dependencies.stream().map(PluginArtifact::fromCoordinates).toList();
+        } catch (IllegalArgumentException e) {
+            stdErr(e.getMessage());
+            return CommandLine.ExitCode.USAGE;
+        }
+
+        final PluginManager pluginManager;
+
+        // If a PLUGIN_PATH is provided, then use the LocalPluginManager
+        if (pluginsPath != null) {
+            pluginManager = new LocalPluginManager(mavenPluginRepositoryProvider.get());
+        } else {
+            // Otherwise, we delegate to the configured plugin-manager.
+            pluginManager = this.pluginManagerProvider.get();
+        }
+
+        List<PluginArtifact> uninstalled = pluginManager.uninstall(
+            pluginArtifacts,
+            false,
+            pluginsPath
+        );
+
+        List<URI> uris = uninstalled.stream().map(PluginArtifact::uri).toList();
+        stdOut("Successfully uninstalled plugins {0} from {1}", dependencies, uris);
+        return CommandLine.ExitCode.OK;
+    }
+
+    @Override
+    protected boolean loadExternalPlugins() {
+        return false;
+    }
+}

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

@@ -1,12 +1,20 @@
 package io.kestra.cli.commands.servers;
 
 import io.kestra.cli.AbstractCommand;
+import io.kestra.core.contexts.KestraContext;
+import jakarta.annotation.PostConstruct;
 import picocli.CommandLine;
 
 abstract public class AbstractServerCommand extends AbstractCommand implements ServerCommandInterface {
     @CommandLine.Option(names = {"--port"}, description = "The port to bind")
     Integer serverPort;
 
+    @Override
+    public Integer call()  throws Exception {
+        this.shutdownHook(true, () -> KestraContext.getContext().shutdown());
+        return super.call();
+    }
+
     protected static int defaultWorkerThread() {
         return Runtime.getRuntime().availableProcessors() * 4;
     }

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

@@ -1,7 +1,6 @@
 package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
-import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.ExecutorInterface;
 import io.kestra.core.services.SkipExecutionService;
@@ -9,7 +8,6 @@ import io.kestra.core.services.StartExecutorService;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
-import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
 import java.util.Collections;
@@ -20,7 +18,6 @@ import java.util.Map;
     name = "executor",
     description = "Start the Kestra executor"
 )
-@Slf4j
 public class ExecutorCommand extends AbstractServerCommand {
     @Inject
     private ApplicationContext applicationContext;
@@ -66,13 +63,10 @@ public class ExecutorCommand extends AbstractServerCommand {
         this.startExecutorService.applyOptions(startExecutors, notStartExecutors);
 
         super.call();
-        this.shutdownHook(() -> KestraContext.getContext().shutdown());
 
         ExecutorInterface executorService = applicationContext.getBean(ExecutorInterface.class);
         executorService.run();
 
-        log.info("Executor started");
-
         Await.until(() -> !this.applicationContext.isRunning());
 
         return 0;

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

@@ -1,13 +1,11 @@
 package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
-import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.runners.IndexerInterface;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
-import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
 import java.util.Map;
@@ -16,7 +14,6 @@ import java.util.Map;
     name = "indexer",
     description = "Start the Kestra indexer"
 )
-@Slf4j
 public class IndexerCommand extends AbstractServerCommand {
     @Inject
     private ApplicationContext applicationContext;
@@ -31,13 +28,10 @@ public class IndexerCommand extends AbstractServerCommand {
     @Override
     public Integer call() throws Exception {
         super.call();
-        this.shutdownHook(() -> KestraContext.getContext().shutdown());
 
         IndexerInterface indexer = applicationContext.getBean(IndexerInterface.class);
         indexer.run();
 
-        log.info("Indexer started");
-
         Await.until(() -> !this.applicationContext.isRunning());
 
         return 0;

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

@@ -1,7 +1,6 @@
 package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
-import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
 import io.kestra.core.schedulers.AbstractScheduler;
 import io.kestra.core.utils.Await;
@@ -31,12 +30,10 @@ public class SchedulerCommand extends AbstractServerCommand {
     @Override
     public Integer call() throws Exception {
         super.call();
-        this.shutdownHook(() -> KestraContext.getContext().shutdown());
 
         AbstractScheduler scheduler = applicationContext.getBean(AbstractScheduler.class);
         scheduler.run();
 
-        log.info("Scheduler started");
         Await.until(() -> !this.applicationContext.isRunning());
 
         return 0;

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

@@ -12,7 +12,6 @@ import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.annotation.Nullable;
 import jakarta.inject.Inject;
-import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
 import java.io.File;
@@ -25,7 +24,6 @@ import java.util.Map;
     name = "standalone",
     description = "Start the standalone all-in-one server"
 )
-@Slf4j
 public class StandAloneCommand extends AbstractServerCommand {
     @CommandLine.Spec
     CommandLine.Model.CommandSpec spec;
@@ -91,11 +89,11 @@ public class StandAloneCommand extends AbstractServerCommand {
         this.skipExecutionService.setSkipFlows(skipFlows);
         this.skipExecutionService.setSkipNamespaces(skipNamespaces);
         this.skipExecutionService.setSkipTenants(skipTenants);
-
         this.startExecutorService.applyOptions(startExecutors, notStartExecutors);
 
+        KestraContext.getContext().injectWorkerConfigs(workerThread, null);
+
         super.call();
-        this.shutdownHook(() -> KestraContext.getContext().shutdown());
 
         if (flowPath != null) {
             try {
@@ -124,8 +122,6 @@ public class StandAloneCommand extends AbstractServerCommand {
             fileWatcher.startListeningFromConfig();
         }
 
-        this.shutdownHook(standAloneRunner::close);
-
         Await.until(() -> !this.applicationContext.isRunning());
 
         return 0;

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

@@ -1,9 +1,7 @@
 package io.kestra.cli.commands.servers;
 
 import com.google.common.collect.ImmutableMap;
-import io.kestra.core.contexts.KestraContext;
 import io.kestra.core.models.ServerType;
-import io.kestra.core.runners.ExecutorInterface;
 import io.kestra.core.runners.IndexerInterface;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
@@ -57,20 +55,11 @@ public class WebServerCommand extends AbstractServerCommand {
             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));
+            shutdownHook(false, () -> poolExecutor.shutdown());
         }
 
         log.info("Webserver started");
-        this.shutdownHook(() -> {
-            this.close();
-            KestraContext.getContext().shutdown();
-        });
         Await.until(() -> !this.applicationContext.isRunning());
         return 0;
     }
-
-    private void close() {
-        if (this.poolExecutor != null) {
-            this.poolExecutor.shutdown();
-        }
-    }
 }

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

@@ -7,7 +7,6 @@ import io.kestra.core.runners.Worker;
 import io.kestra.core.utils.Await;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
-import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 import picocli.CommandLine.Option;
 
@@ -18,7 +17,6 @@ import java.util.UUID;
     name = "worker",
     description = "Start the Kestra worker"
 )
-@Slf4j
 public class WorkerCommand extends AbstractServerCommand {
 
     @Inject
@@ -39,8 +37,11 @@ public class WorkerCommand extends AbstractServerCommand {
 
     @Override
     public Integer call() throws Exception {
+
+        KestraContext.getContext().injectWorkerConfigs(thread, workerGroupKey);
+
         super.call();
-        this.shutdownHook(() -> KestraContext.getContext().shutdown());
+
         if (this.workerGroupKey != null && !this.workerGroupKey.matches("[a-zA-Z0-9_-]+")) {
             throw new IllegalArgumentException("The --worker-group option must match the [a-zA-Z0-9_-]+ pattern");
         }
@@ -52,13 +53,6 @@ public class WorkerCommand extends AbstractServerCommand {
 
         worker.run();
 
-        if (this.workerGroupKey != null) {
-            log.info("Worker started with {} thread(s) in group '{}'", this.thread, this.workerGroupKey);
-        }
-        else {
-            log.info("Worker started with {} thread(s)", this.thread);
-        }
-
         Await.until(() -> !this.applicationContext.isRunning());
 
         return 0;

cli/src/main/java/io/kestra/cli/commands/sys/ReindexCommand.java

@@ -2,6 +2,7 @@ package io.kestra.cli.commands.sys;
 
 import io.kestra.cli.AbstractCommand;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
@@ -9,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
 import java.util.List;
+import java.util.Objects;
 
 @CommandLine.Command(
     name = "reindex",
@@ -33,8 +35,8 @@ public class ReindexCommand extends AbstractCommand {
             List<Flow> allFlow = flowRepository.findAllForAllTenants();
             allFlow.stream()
                 .map(flow -> flowRepository.findByIdWithSource(flow.getTenantId(), flow.getNamespace(), flow.getId()).orElse(null))
-                .filter(flow -> flow != null)
-                .forEach(flow -> flowRepository.update(flow.toFlow(), flow.toFlow(), flow.getSource(), flow.toFlow()));
+                .filter(Objects::nonNull)
+                .forEach(flow -> flowRepository.update(GenericFlow.of(flow), flow));
 
             stdOut("Successfully reindex " + allFlow.size() + " flow(s).");
         }

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

@@ -4,7 +4,6 @@ import io.kestra.cli.AbstractValidateCommand;
 import io.kestra.core.models.templates.Template;
 import io.kestra.core.models.templates.TemplateEnabled;
 import io.kestra.core.models.validations.ModelValidator;
-import io.kestra.core.serializers.YamlParser;
 import jakarta.inject.Inject;
 import picocli.CommandLine;
 
@@ -16,8 +15,6 @@ import java.util.Collections;
 )
 @TemplateEnabled
 public class TemplateValidateCommand extends AbstractValidateCommand {
-    @Inject
-    private YamlParser yamlParser;
 
     @Inject
     private ModelValidator modelValidator;
@@ -26,7 +23,6 @@ public class TemplateValidateCommand extends AbstractValidateCommand {
     public Integer call() throws Exception {
         return this.call(
             Template.class,
-            yamlParser,
             modelValidator,
             (Object object) -> {
                 Template template = (Template) object;

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

@@ -10,7 +10,6 @@ import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MutableHttpRequest;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.http.client.netty.DefaultHttpClient;
-import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine;
 
@@ -27,8 +26,6 @@ import jakarta.validation.ConstraintViolationException;
 @Slf4j
 @TemplateEnabled
 public class TemplateNamespaceUpdateCommand extends AbstractServiceNamespaceUpdateCommand {
-    @Inject
-    public YamlParser yamlParser;
 
     @Override
     public Integer call() throws Exception {
@@ -38,7 +35,7 @@ public class TemplateNamespaceUpdateCommand extends AbstractServiceNamespaceUpda
             List<Template> templates = files
                 .filter(Files::isRegularFile)
                 .filter(YamlParser::isValidExtension)
-                .map(path -> yamlParser.parse(path.toFile(), Template.class))
+                .map(path -> YamlParser.parse(path.toFile(), Template.class))
                 .toList();
 
             if (templates.isEmpty()) {

cli/src/main/java/io/kestra/cli/plugins/PluginDownloader.java

@@ -1,153 +0,0 @@
-package io.kestra.cli.plugins;
-
-import com.google.common.collect.ImmutableList;
-import io.micronaut.context.annotation.Value;
-import io.micronaut.core.annotation.Nullable;
-import jakarta.inject.Inject;
-import jakarta.inject.Singleton;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.io.FileUtils;
-import org.apache.maven.repository.internal.MavenRepositorySystemUtils;
-import org.eclipse.aether.DefaultRepositorySystemSession;
-import org.eclipse.aether.RepositorySystem;
-import org.eclipse.aether.RepositorySystemSession;
-import org.eclipse.aether.artifact.Artifact;
-import org.eclipse.aether.artifact.DefaultArtifact;
-import org.eclipse.aether.connector.basic.BasicRepositoryConnectorFactory;
-import org.eclipse.aether.impl.DefaultServiceLocator;
-import org.eclipse.aether.repository.LocalRepository;
-import org.eclipse.aether.repository.RemoteRepository;
-import org.eclipse.aether.resolution.*;
-import org.eclipse.aether.spi.connector.RepositoryConnectorFactory;
-import org.eclipse.aether.spi.connector.transport.TransporterFactory;
-import org.eclipse.aether.transport.file.FileTransporterFactory;
-import org.eclipse.aether.transport.http.HttpTransporterFactory;
-import org.eclipse.aether.util.repository.AuthenticationBuilder;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.file.Files;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.stream.Collectors;
-
-@Singleton
-@Slf4j
-public class PluginDownloader {
-    private final List<RepositoryConfig> repositoryConfigs;
-    private final RepositorySystem system;
-    private final RepositorySystemSession session;
-
-    @Inject
-    public PluginDownloader(
-        List<RepositoryConfig> repositoryConfigs,
-        @Nullable @Value("${kestra.plugins.local-repository-path}") String localRepositoryPath
-    ) {
-        this.repositoryConfigs = repositoryConfigs;
-        this.system = repositorySystem();
-        this.session = repositorySystemSession(system, localRepositoryPath);
-    }
-
-    public void addRepository(RepositoryConfig repositoryConfig) {
-        this.repositoryConfigs.add(repositoryConfig);
-    }
-
-    public List<URL> resolve(List<String> dependencies) throws MalformedURLException, ArtifactResolutionException, VersionRangeResolutionException {
-        List<RemoteRepository> repositories = remoteRepositories();
-
-        List<ArtifactResult> artifactResults = resolveArtifacts(repositories, dependencies);
-        List<URL> localUrls = resolveUrls(artifactResults);
-        log.debug("Resolved Plugin {} with {}", dependencies, localUrls);
-
-        return localUrls;
-    }
-
-    private List<RemoteRepository> remoteRepositories() {
-        return repositoryConfigs
-            .stream()
-            .map(repositoryConfig -> {
-                var build = new RemoteRepository.Builder(
-                    repositoryConfig.getId(),
-                    "default",
-                    repositoryConfig.getUrl()
-                );
-
-                if (repositoryConfig.getBasicAuth() != null) {
-                    var authenticationBuilder = new AuthenticationBuilder();
-                    authenticationBuilder.addUsername(repositoryConfig.getBasicAuth().getUsername());
-                    authenticationBuilder.addPassword(repositoryConfig.getBasicAuth().getPassword());
-
-                    build.setAuthentication(authenticationBuilder.build());
-                }
-
-                return build.build();
-            })
-            .toList();
-    }
-
-    private static RepositorySystem repositorySystem() {
-        DefaultServiceLocator locator = MavenRepositorySystemUtils.newServiceLocator();
-        locator.addService(RepositoryConnectorFactory.class, BasicRepositoryConnectorFactory.class);
-        locator.addService(TransporterFactory.class, FileTransporterFactory.class);
-        locator.addService(TransporterFactory.class, HttpTransporterFactory.class);
-
-        return locator.getService(RepositorySystem.class);
-    }
-
-    private RepositorySystemSession repositorySystemSession(RepositorySystem system, String localRepositoryPath) {
-        DefaultRepositorySystemSession session = MavenRepositorySystemUtils.newSession();
-
-        if (localRepositoryPath == null) {
-            try {
-                final String tempDirectory = Files.createTempDirectory(this.getClass().getSimpleName().toLowerCase())
-                    .toAbsolutePath()
-                    .toString();
-
-                localRepositoryPath = tempDirectory;
-
-                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-                    try {
-                        FileUtils.deleteDirectory(new File(tempDirectory));
-                    } catch (IOException e) {
-                        throw new RuntimeException(e);
-                    }
-                }));
-            } catch (IOException e) {
-                throw new RuntimeException(e);
-            }
-        }
-
-        LocalRepository localRepo = new LocalRepository(localRepositoryPath);
-        session.setLocalRepositoryManager(system.newLocalRepositoryManager(session, localRepo));
-
-        return session;
-    }
-
-    private List<ArtifactResult> resolveArtifacts(List<RemoteRepository> repositories, List<String> dependencies) throws ArtifactResolutionException, VersionRangeResolutionException {
-        List<ArtifactResult> results = new ArrayList<>(dependencies.size());
-        for (String dependency: dependencies) {
-            var artifact = new DefaultArtifact(dependency);
-            var version = system.resolveVersionRange(session, new VersionRangeRequest(artifact, repositories, null));
-            var artifactRequest = new ArtifactRequest(
-                new DefaultArtifact(artifact.getGroupId(), artifact.getArtifactId(), "jar", version.getHighestVersion().toString()),
-                repositories,
-                null
-            );
-            var artifactResult = system.resolveArtifact(session, artifactRequest);
-            results.add(artifactResult);
-        }
-        return results;
-    }
-
-    private List<URL> resolveUrls(List<ArtifactResult> artifactResults) throws MalformedURLException {
-        ImmutableList.Builder<URL> urls = ImmutableList.builder();
-        for (ArtifactResult artifactResult : artifactResults) {
-            URL url;
-            url = artifactResult.getArtifact().getFile().toPath().toUri().toURL();
-            urls.add(url);
-        }
-        return urls.build();
-    }
-}

cli/src/main/java/io/kestra/cli/plugins/RepositoryConfig.java

@@ -1,30 +0,0 @@
-package io.kestra.cli.plugins;
-
-import io.micronaut.context.annotation.EachProperty;
-import io.micronaut.context.annotation.Parameter;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-
-@EachProperty("kestra.plugins.repositories")
-@Getter
-@AllArgsConstructor
-@Builder
-public class RepositoryConfig {
-    String id;
-
-    String url;
-
-    BasicAuth basicAuth;
-
-    @Getter
-    @AllArgsConstructor
-    public static class BasicAuth {
-        private String username;
-        private String password;
-    }
-
-    public RepositoryConfig(@Parameter String id) {
-        this.id = id;
-    }
-}

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

@@ -1,22 +1,23 @@
 package io.kestra.cli.services;
 
-import io.kestra.core.models.flows.Flow;
+import io.kestra.core.exceptions.DeserializationException;
+import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.FlowWithPath;
 import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.models.validations.ModelValidator;
 import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.serializers.YamlParser;
 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.inject.Inject;
+import jakarta.annotation.Nullable;
 import jakarta.inject.Singleton;
 import jakarta.validation.ConstraintViolationException;
 import lombok.extern.slf4j.Slf4j;
 
-import javax.annotation.Nullable;
 import java.io.IOException;
 import java.nio.charset.Charset;
 import java.nio.file.*;
@@ -40,9 +41,6 @@ public class FileChangedEventListener {
     @Inject
     private PluginDefaultService pluginDefaultService;
 
-    @Inject
-    private YamlParser yamlParser;
-
     @Inject
     private ModelValidator modelValidator;
 
@@ -59,7 +57,6 @@ public class FileChangedEventListener {
 
     private boolean isStarted = false;
 
-
     @Inject
     public FileChangedEventListener(@Nullable FileWatchConfiguration fileWatchConfiguration, @Nullable WatchService watchService) {
         this.fileWatchConfiguration = fileWatchConfiguration;
@@ -68,7 +65,7 @@ public class FileChangedEventListener {
 
     public void startListeningFromConfig() throws IOException, InterruptedException {
         if (fileWatchConfiguration != null && fileWatchConfiguration.isEnabled()) {
-            this.flowFilesManager = new LocalFlowFileWatcher(flowRepositoryInterface, pluginDefaultService);
+            this.flowFilesManager = new LocalFlowFileWatcher(flowRepositoryInterface);
             List<Path> paths = fileWatchConfiguration.getPaths();
             this.setup(paths);
 
@@ -76,7 +73,7 @@ public class FileChangedEventListener {
             // Init existing flows not already in files
             flowListeners.listen(flows -> {
                 if (!isStarted) {
-                    for (FlowWithSource flow : flows) {
+                    for (FlowInterface flow : flows) {
                         if (this.flows.stream().noneMatch(flowWithPath -> flowWithPath.uidWithoutRevision().equals(flow.uidWithoutRevision()))) {
                             flowToFile(flow, this.buildPath(flow));
                             this.flows.add(FlowWithPath.of(flow, this.buildPath(flow).toString()));
@@ -137,7 +134,7 @@ public class FileChangedEventListener {
                                 try {
                                     String content = Files.readString(filePath, Charset.defaultCharset());
 
-                                    Optional<Flow> flow = parseFlow(content, entry);
+                                    Optional<FlowWithSource> flow = parseFlow(content, entry);
                                     if (flow.isPresent()) {
                                         if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
                                             // Check if we already have a file with the given path
@@ -156,7 +153,7 @@ public class FileChangedEventListener {
                                             flows.add(FlowWithPath.of(flow.get(), filePath.toString()));
                                         }
 
-                                        flowFilesManager.createOrUpdateFlow(flow.get(), content);
+                                        flowFilesManager.createOrUpdateFlow(GenericFlow.fromYaml(tenantId, content));
                                         log.info("Flow {} from file {} has been created or modified", flow.get().getId(), entry);
                                     }
 
@@ -207,11 +204,11 @@ public class FileChangedEventListener {
                 public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                     if (file.toString().endsWith(".yml") || file.toString().endsWith(".yaml")) {
                         String content = Files.readString(file, Charset.defaultCharset());
-                        Optional<Flow> flow = parseFlow(content, file);
+                        Optional<FlowWithSource> flow = parseFlow(content, file);
 
                         if (flow.isPresent() && flows.stream().noneMatch(flowWithPath -> flowWithPath.uidWithoutRevision().equals(flow.get().uidWithoutRevision()))) {
                             flows.add(FlowWithPath.of(flow.get(), file.toString()));
-                            flowFilesManager.createOrUpdateFlow(flow.get(), content);
+                            flowFilesManager.createOrUpdateFlow(GenericFlow.fromYaml(tenantId, content));
                         }
                     }
                     return FileVisitResult.CONTINUE;
@@ -223,27 +220,25 @@ public class FileChangedEventListener {
         }
     }
 
-    private void flowToFile(FlowWithSource flow, Path path) {
+    private void flowToFile(FlowInterface flow, Path path) {
         Path defaultPath = path != null ? path : this.buildPath(flow);
 
         try {
-            Files.writeString(defaultPath, flow.getSource());
+            Files.writeString(defaultPath, flow.source());
             log.info("Flow {} has been written to file {}", flow.getId(), defaultPath);
         } catch (IOException e) {
             log.error("Error writing file: {}", defaultPath, e);
         }
     }
 
-    private Optional<Flow> parseFlow(String content, Path entry) {
+    private Optional<FlowWithSource> parseFlow(String content, Path entry) {
         try {
-            Flow flow = yamlParser.parse(content, Flow.class);
-            FlowWithSource withPluginDefault = pluginDefaultService.injectDefaults(FlowWithSource.of(flow, content));
-            modelValidator.validate(withPluginDefault);
+            FlowWithSource flow = pluginDefaultService.parseFlowWithAllDefaults(tenantId, content, false);
+            modelValidator.validate(flow);
             return Optional.of(flow);
-        } catch (ConstraintViolationException e) {
+        } catch (DeserializationException | ConstraintViolationException e) {
             log.warn("Error while parsing flow: {}", entry, e);
         }
-
         return Optional.empty();
     }
 
@@ -259,7 +254,7 @@ public class FileChangedEventListener {
         }
     }
 
-    private Path buildPath(Flow flow) {
+    private Path buildPath(FlowInterface flow) {
         return fileWatchConfiguration.getPaths().getFirst().resolve(flow.uidWithoutRevision() + ".yml");
     }
 }

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

@@ -1,11 +1,11 @@
 package io.kestra.cli.services;
 
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.flows.GenericFlow;
 
 public interface FlowFilesManager {
 
-    FlowWithSource createOrUpdateFlow(Flow flow, String content);
+    FlowWithSource createOrUpdateFlow(GenericFlow flow);
 
     void deleteFlow(FlowWithSource toDelete);
 

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

@@ -1,27 +1,23 @@
 package io.kestra.cli.services;
 
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.services.PluginDefaultService;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class LocalFlowFileWatcher implements FlowFilesManager {
     private final FlowRepositoryInterface flowRepository;
-    private final PluginDefaultService pluginDefaultService;
 
-    public LocalFlowFileWatcher(FlowRepositoryInterface flowRepository, PluginDefaultService pluginDefaultService) {
+    public LocalFlowFileWatcher(FlowRepositoryInterface flowRepository) {
         this.flowRepository = flowRepository;
-        this.pluginDefaultService = pluginDefaultService;
     }
 
     @Override
-    public FlowWithSource createOrUpdateFlow(Flow flow, String content) {
-        FlowWithSource withDefault = pluginDefaultService.injectDefaults(FlowWithSource.of(flow, content));
+    public FlowWithSource createOrUpdateFlow(final GenericFlow flow) {
         return flowRepository.findById(null, flow.getNamespace(), flow.getId())
-            .map(previous -> flowRepository.update(flow, previous, content, withDefault))
-            .orElseGet(() -> flowRepository.create(flow, content, withDefault));
+            .map(previous -> flowRepository.update(flow, previous))
+            .orElseGet(() -> flowRepository.create(flow));
     }
 
     @Override

cli/src/main/resources/application.yml

@@ -1,6 +1,9 @@
 micronaut:
   application:
     name: kestra
+  # Disable Micronaut Open Telemetry
+  otel:
+    enabled: false
   router:
     static-resources:
       swagger:

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

@@ -32,6 +32,8 @@ class FlowExportCommandTest {
 
             // we use the update command to add flows to extract
             String[] updateArgs = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -44,6 +46,8 @@ class FlowExportCommandTest {
 
             // then we export them
             String[] exportArgs = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",

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

@@ -28,6 +28,8 @@ class FlowUpdatesCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -41,6 +43,8 @@ class FlowUpdatesCommandTest {
             out.reset();
 
             args = new String[]{
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -70,6 +74,8 @@ class FlowUpdatesCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -84,6 +90,8 @@ class FlowUpdatesCommandTest {
 
             // no "delete" arg should behave as no-delete
             args = new String[]{
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -96,6 +104,8 @@ class FlowUpdatesCommandTest {
             out.reset();
 
             args = new String[]{
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -109,6 +119,35 @@ class FlowUpdatesCommandTest {
         }
     }
 
+    @Test
+    void invalidWithNamespace() {
+        URL directory = FlowUpdatesCommandTest.class.getClassLoader().getResource("flows");
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setErr(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+
+            EmbeddedServer embeddedServer = ctx.getBean(EmbeddedServer.class);
+            embeddedServer.start();
+
+            String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
+                "--server",
+                embeddedServer.getURL().toString(),
+                "--user",
+                "myuser:pass:word",
+                "--namespace",
+                "io.kestra.cli",
+                "--delete",
+                directory.getPath(),
+            };
+            PicocliRunner.call(FlowUpdatesCommand.class, ctx, args);
+
+            assertThat(out.toString(), containsString("Invalid entity: flow.namespace: io.kestra.outsider_quattro_-1 - flow namespace is invalid"));
+        }
+    }
+
     @Test
     void helper()  {
         URL directory = FlowUpdatesCommandTest.class.getClassLoader().getResource("helper");
@@ -121,6 +160,8 @@ class FlowUpdatesCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",

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

@@ -46,6 +46,8 @@ class TemplateValidateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",

cli/src/test/java/io/kestra/cli/commands/namespaces/files/NamespaceFilesUpdateCommandTest.java

@@ -31,6 +31,8 @@ class NamespaceFilesUpdateCommandTest {
 
             String to = "/some/directory";
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -61,6 +63,8 @@ class NamespaceFilesUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -90,6 +94,8 @@ class NamespaceFilesUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",

cli/src/test/java/io/kestra/cli/commands/namespaces/kv/KvUpdateCommandTest.java

@@ -28,6 +28,8 @@ class KvUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -54,6 +56,8 @@ class KvUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -80,6 +84,8 @@ class KvUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -108,6 +114,8 @@ class KvUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -134,6 +142,8 @@ class KvUpdateCommandTest {
             embeddedServer.start();
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
@@ -167,6 +177,8 @@ class KvUpdateCommandTest {
             Files.write(file.toPath(), "{\"some\":\"json\",\"from\":\"file\"}".getBytes());
 
             String[] args = {
+                "--plugins",
+                "/tmp", // pass this arg because it can cause failure
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",

cli/src/test/java/io/kestra/cli/commands/plugins/PluginCommandTest.java

@@ -0,0 +1,27 @@
+package io.kestra.cli.commands.plugins;
+
+import io.micronaut.configuration.picocli.PicocliRunner;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.env.Environment;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.StringContains.containsString;
+
+class PluginCommandTest {
+
+    @Test
+    void shouldGetHelps() {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+            PicocliRunner.call(PluginCommand.class, ctx);
+
+            assertThat(out.toString(), containsString("Usage: kestra plugins"));
+        }
+    }
+}

cli/src/test/java/io/kestra/cli/commands/plugins/PluginInstallCommandTest.java

@@ -9,7 +9,6 @@ import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
@@ -17,7 +16,7 @@ import static org.hamcrest.Matchers.*;
 class PluginInstallCommandTest {
 
     @Test
-    void fixedVersion() throws IOException {
+    void shouldInstallPluginLocallyGivenFixedVersion() throws IOException {
         Path pluginsPath = Files.createTempDirectory(PluginInstallCommandTest.class.getSimpleName());
         pluginsPath.toFile().deleteOnExit();
 
@@ -28,12 +27,12 @@ class PluginInstallCommandTest {
             List<Path> files = Files.list(pluginsPath).toList();
 
             assertThat(files.size(), is(1));
-            assertThat(files.getFirst().getFileName().toString(), is("plugin-notifications-0.6.0.jar"));
+            assertThat(files.getFirst().getFileName().toString(), is("io_kestra_plugin__plugin-notifications__0_6_0.jar"));
         }
     }
 
     @Test
-    void latestVersion() throws IOException {
+    void shouldInstallPluginLocallyGivenLatestVersion() throws IOException {
         Path pluginsPath = Files.createTempDirectory(PluginInstallCommandTest.class.getSimpleName());
         pluginsPath.toFile().deleteOnExit();
 
@@ -44,13 +43,13 @@ class PluginInstallCommandTest {
             List<Path> files = Files.list(pluginsPath).toList();
 
             assertThat(files.size(), is(1));
-            assertThat(files.getFirst().getFileName().toString(), startsWith("plugin-notifications"));
+            assertThat(files.getFirst().getFileName().toString(), startsWith("io_kestra_plugin__plugin-notifications__"));
             assertThat(files.getFirst().getFileName().toString(), not(containsString("LATEST")));
         }
     }
 
     @Test
-    void rangeVersion() throws IOException {
+    void shouldInstallPluginLocallyGivenRangeVersion() throws IOException {
         Path pluginsPath = Files.createTempDirectory(PluginInstallCommandTest.class.getSimpleName());
         pluginsPath.toFile().deleteOnExit();
 
@@ -62,7 +61,7 @@ class PluginInstallCommandTest {
             List<Path> files = Files.list(pluginsPath).toList();
 
             assertThat(files.size(), is(1));
-            assertThat(files.getFirst().getFileName().toString(), is("storage-s3-0.12.1.jar"));
+            assertThat(files.getFirst().getFileName().toString(), is("io_kestra_storage__storage-s3__0_12_1.jar"));
         }
     }
 }

cli/src/test/java/io/kestra/cli/commands/plugins/PluginListCommandTest.java

@@ -4,7 +4,6 @@ import io.micronaut.configuration.picocli.PicocliRunner;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.env.Environment;
 import org.apache.commons.io.FileUtils;
-import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 import java.io.ByteArrayOutputStream;
@@ -25,7 +24,7 @@ class PluginListCommandTest {
     private static final String PLUGIN_TEMPLATE_TEST = "plugin-template-test-0.18.0-SNAPSHOT.jar";
 
     @Test
-    void run() throws IOException, URISyntaxException {
+    void shouldListPluginsInstalledLocally() throws IOException, URISyntaxException {
         Path pluginsPath = Files.createTempDirectory(PluginListCommandTest.class.getSimpleName());
         pluginsPath.toFile().deleteOnExit();
 

cli/src/test/java/io/kestra/cli/commands/plugins/PluginSearchCommandTest.java

@@ -0,0 +1,105 @@
+package io.kestra.cli.commands.plugins;
+
+import com.github.tomakehurst.wiremock.junit5.WireMockTest;
+import io.micronaut.configuration.picocli.PicocliRunner;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.env.Environment;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.Map;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+@WireMockTest(httpPort = 28181)
+class PluginSearchCommandTest {
+    private ByteArrayOutputStream outputStreamCaptor;
+    private final PrintStream originalOut = System.out;
+
+    @BeforeEach
+    void setUp() {
+        outputStreamCaptor = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(outputStreamCaptor));
+    }
+
+    @AfterEach
+    void tearDown() {
+        System.setOut(originalOut);
+    }
+
+    @Test
+    void searchWithExactMatch() {
+        stubFor(get(urlEqualTo("/v1/plugins"))
+            .willReturn(aResponse()
+                .withHeader("Content-Type", "application/json")
+                .withBody("""
+                    [
+                        {
+                            "name": "plugin-notifications",
+                            "title": "Notifications",
+                            "group": "io.kestra.plugin",
+                            "version": "0.6.0"
+                        },
+                        {
+                            "name": "plugin-scripts",
+                            "title": "Scripts",
+                            "group": "io.kestra.plugin",
+                            "version": "0.5.0"
+                        }
+                    ]
+                """)));
+
+        try (ApplicationContext ctx = ApplicationContext.builder(Environment.CLI, Environment.TEST)
+            .properties(Map.of("micronaut.http.services.api.url", "http://localhost:28181"))
+            .start()) {
+            String[] args = {"notifications"};
+            PicocliRunner.call(PluginSearchCommand.class, ctx, args);
+
+            String output = outputStreamCaptor.toString().trim();
+            assertThat(output, containsString("Found 1 plugins matching 'notifications'"));
+            assertThat(output, containsString("plugin-notifications"));
+            assertThat(output, not(containsString("plugin-scripts")));
+        }
+    }
+
+    @Test
+    void searchWithEmptyQuery() {
+        stubFor(get(urlEqualTo("/v1/plugins"))
+            .willReturn(aResponse()
+                .withHeader("Content-Type", "application/json")
+                .withBody("""
+                    [
+                        {
+                            "name": "plugin-notifications",
+                            "title": "Notifications",
+                            "group": "io.kestra.plugin",
+                            "version": "0.6.0"
+                        },
+                        {
+                            "name": "plugin-scripts",
+                            "title": "Scripts",
+                            "group": "io.kestra.plugin",
+                            "version": "0.5.0"
+                        }
+                    ]
+                """)));
+
+        try (ApplicationContext ctx = ApplicationContext.builder(Environment.CLI, Environment.TEST)
+            .properties(Map.of("micronaut.http.services.api.url", "http://localhost:28181"))
+            .start()) {
+
+            String[] args = {""};
+            PicocliRunner.call(PluginSearchCommand.class, ctx, args);
+
+            String output = outputStreamCaptor.toString().trim();
+            assertThat(output, containsString("Found 2 plugins"));
+            assertThat(output, containsString("plugin-notifications"));
+            assertThat(output, containsString("plugin-scripts"));
+        }
+    }
+}

cli/src/test/java/io/kestra/cli/commands/sys/statestore/StateStoreMigrateCommandTest.java

@@ -1,16 +1,15 @@
 package io.kestra.cli.commands.sys.statestore;
 
-import com.devskiller.friendly_id.FriendlyId;
 import io.kestra.core.exceptions.MigrationRequiredException;
 import io.kestra.core.exceptions.ResourceExpiredException;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.storages.StateStore;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.Hashing;
-import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.Slugify;
 import io.kestra.plugin.core.log.Log;
 import io.micronaut.configuration.picocli.PicocliRunner;
@@ -27,7 +26,6 @@ import java.util.List;
 import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.core.Is.is;
 
 class StateStoreMigrateCommandTest {
@@ -45,7 +43,7 @@ class StateStoreMigrateCommandTest {
                 .namespace("some.valid.namespace." + ((int) (Math.random() * 1000000)))
                 .tasks(List.of(Log.builder().id("log").type(Log.class.getName()).message("logging").build()))
                 .build();
-            flowRepository.create(flow, flow.generateSource(), flow);
+            flowRepository.create(GenericFlow.of(flow));
 
             StorageInterface storage = ctx.getBean(StorageInterface.class);
             String tenantId = flow.getTenantId();

core/build.gradle

@@ -38,6 +38,13 @@ dependencies {
     implementation group: 'dev.failsafe', name: 'failsafe'
     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-connector-basic'
+    implementation 'org.apache.maven.resolver:maven-resolver-transport-file'
+    implementation 'org.apache.maven.resolver:maven-resolver-transport-http'
+
     // scheduler
     implementation group: 'com.cronutils', name: 'cron-utils'
 
@@ -66,7 +73,7 @@ dependencies {
     testImplementation "io.micronaut:micronaut-http-server-netty"
     testImplementation "io.micronaut:micronaut-management"
 
-    testImplementation "org.testcontainers:testcontainers:1.20.4"
-    testImplementation "org.testcontainers:junit-jupiter:1.20.4"
+    testImplementation "org.testcontainers:testcontainers:1.20.6"
+    testImplementation "org.testcontainers:junit-jupiter:1.20.6"
     testImplementation "org.bouncycastle:bcpkix-jdk18on:1.80"
 }

core/src/main/java/io/kestra/core/contexts/KestraBeansFactory.java

@@ -3,7 +3,6 @@ package io.kestra.core.contexts;
 import io.kestra.core.exceptions.KestraRuntimeException;
 import io.kestra.core.plugins.DefaultPluginRegistry;
 import io.kestra.core.plugins.PluginRegistry;
-import io.kestra.core.plugins.serdes.PluginDeserializer;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.storages.StorageInterfaceFactory;
 import io.micronaut.context.annotation.Bean;
@@ -34,7 +33,7 @@ public class KestraBeansFactory {
     StorageConfig storageConfig;
 
     @Value("${kestra.storage.type}")
-    Optional<String> storageType;
+    protected Optional<String> storageType;
 
     @Requires(missingBeans = PluginRegistry.class)
     @Singleton
@@ -42,16 +41,25 @@ public class KestraBeansFactory {
         return DefaultPluginRegistry.getOrCreate();
     }
 
+    @Singleton
+    public StorageInterfaceFactory storageInterfaceFactory(final PluginRegistry pluginRegistry){
+        return new StorageInterfaceFactory(pluginRegistry, validator);
+    }
+
     @Requires(missingBeans = StorageInterface.class)
     @Singleton
     @Bean(preDestroy = "close")
-    public StorageInterface storageInterface(final PluginRegistry pluginRegistry) throws IOException {
-        String pluginId = storageType.orElseThrow(() -> new KestraRuntimeException(String.format(
+    public StorageInterface storageInterface(final StorageInterfaceFactory storageInterfaceFactory) throws IOException {
+        String pluginId = getStoragePluginId(storageInterfaceFactory);
+        return storageInterfaceFactory.make(null, pluginId, storageConfig.getStorageConfig(pluginId));
+    }
+
+    public String getStoragePluginId(StorageInterfaceFactory storageInterfaceFactory) {
+        return storageType.orElseThrow(() -> new KestraRuntimeException(String.format(
             "No storage configured through the application property '%s'. Supported types are: %s"
             , KESTRA_STORAGE_TYPE_CONFIG,
-            StorageInterfaceFactory.getLoggableStorageIds(pluginRegistry)
+            storageInterfaceFactory.getLoggableStorageIds()
         )));
-        return StorageInterfaceFactory.make(pluginRegistry, pluginId, storageConfig.getStorageConfig(pluginId), validator);
     }
 
     @ConfigurationProperties("kestra")
@@ -67,7 +75,7 @@ public class KestraBeansFactory {
          * @return the configuration.
          */
         @SuppressWarnings("unchecked")
-        private Map<String, Object> getStorageConfig(String type) {
+        public Map<String, Object> getStorageConfig(String type) {
             return (Map<String, Object>) storage.get(StringConvention.CAMEL_CASE.format(type));
         }
     }

core/src/main/java/io/kestra/core/contexts/KestraContext.java

@@ -10,6 +10,8 @@ import io.micronaut.context.env.Environment;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -25,7 +27,11 @@ public abstract class KestraContext {
     private static final AtomicReference<KestraContext> INSTANCE = new AtomicReference<>();
 
     // Properties
-    private static final String KESTRA_SERVER_TYPE = "kestra.server-type";
+    public static final String KESTRA_SERVER_TYPE = "kestra.server-type";
+
+    // Those properties are injected bases on the CLI args.
+    private static final String KESTRA_WORKER_MAX_NUM_THREADS = "kestra.worker.max-num-threads";
+    private static final String KESTRA_WORKER_GROUP_KEY = "kestra.worker.group-key";
 
     /**
      * Gets the current {@link KestraContext}.
@@ -54,6 +60,12 @@ public abstract class KestraContext {
      */
     public abstract ServerType getServerType();
 
+    public abstract Optional<Integer> getWorkerMaxNumThreads();
+
+    public abstract Optional<String> getWorkerGroupKey();
+
+    public abstract void injectWorkerConfigs(Integer maxNumThreads, String workerGroupKey);
+
     /**
      * Returns the Kestra Version.
      *
@@ -110,6 +122,34 @@ public abstract class KestraContext {
                 .orElse(ServerType.STANDALONE);
         }
 
+        /** {@inheritDoc} **/
+        @Override
+        public Optional<Integer> getWorkerMaxNumThreads() {
+            return Optional.ofNullable(environment)
+                .flatMap(env -> env.getProperty(KESTRA_WORKER_MAX_NUM_THREADS, Integer.class));
+        }
+
+        /** {@inheritDoc} **/
+        @Override
+        public Optional<String> getWorkerGroupKey() {
+            return Optional.ofNullable(environment)
+                .flatMap(env -> env.getProperty(KESTRA_WORKER_GROUP_KEY, String.class));
+        }
+        /** {@inheritDoc} **/
+        @Override
+        public void injectWorkerConfigs(Integer maxNumThreads, String workerGroupKey) {
+            final Map<String, Object> configs = new HashMap<>();
+            Optional.ofNullable(maxNumThreads)
+                .ifPresent(val -> configs.put(KESTRA_WORKER_MAX_NUM_THREADS, val));
+
+            Optional.ofNullable(workerGroupKey)
+                .ifPresent(val -> configs.put(KESTRA_WORKER_GROUP_KEY, val));
+
+            if (!configs.isEmpty()) {
+                environment.addPropertySource("kestra-runtime", configs);
+            }
+        }
+
         /** {@inheritDoc} **/
         @Override
         public void shutdown() {

core/src/main/java/io/kestra/core/contexts/MavenPluginRepositoryConfig.java

@@ -0,0 +1,27 @@
+package io.kestra.core.contexts;
+
+import io.micronaut.context.annotation.ConfigurationProperties;
+import io.micronaut.context.annotation.EachProperty;
+import io.micronaut.context.annotation.Parameter;
+import io.micronaut.core.annotation.Nullable;
+import lombok.Builder;
+
+@Builder
+@EachProperty("kestra.plugins.repositories")
+public record MavenPluginRepositoryConfig(
+    @Parameter
+    String id,
+    String url,
+    @Nullable
+    BasicAuth basicAuth
+) {
+
+    @Builder
+    @ConfigurationProperties("basic-auth")
+    public record BasicAuth(
+        String username,
+        String password
+    ) {
+
+    }
+}

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

@@ -1,6 +1,7 @@
 package io.kestra.core.docs;
 
 import com.google.common.base.CaseFormat;
+import io.kestra.core.models.Plugin;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.models.tasks.runners.TaskRunner;
 import lombok.AllArgsConstructor;

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

@@ -1,10 +1,9 @@
 package io.kestra.core.docs;
 
-import io.kestra.core.plugins.RegisteredPlugin;
+import io.kestra.core.plugins.PluginClassAndMetadata;
 import lombok.*;
 
 import java.util.*;
-import java.util.stream.Collectors;
 
 @Getter
 @EqualsAndHashCode
@@ -21,16 +20,18 @@ public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
     private Map<String, Object> outputsSchema;
 
     @SuppressWarnings("unchecked")
-    private ClassPluginDocumentation(JsonSchemaGenerator jsonSchemaGenerator, RegisteredPlugin plugin, Class<? extends T> cls, Class<T> baseCls, String alias) {
-        super(jsonSchemaGenerator, cls, baseCls);
+    private ClassPluginDocumentation(JsonSchemaGenerator jsonSchemaGenerator, PluginClassAndMetadata<T> plugin, boolean allProperties) {
+        super(jsonSchemaGenerator, plugin.type(), allProperties ? null : plugin.baseClass());
 
         // plugins metadata
-        this.cls = alias == null ? cls.getName() : alias;
+        Class<? extends T> cls = plugin.type();
+
+        this.cls = plugin.alias() == null ? cls.getName() : plugin.alias();
         this.group = plugin.group();
         this.docLicense = plugin.license();
         this.pluginTitle = plugin.title();
-        this.icon = plugin.icon(cls);
-        if (alias != null) {
+        this.icon = plugin.icon();
+        if (plugin.alias() != null) {
             replacement = cls.getName();
         }
 
@@ -38,10 +39,10 @@ public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
             this.subGroup = cls.getPackageName().substring(this.group.length() + 1);
         }
 
-        this.shortName = alias == null ? cls.getSimpleName() : alias.substring(alias.lastIndexOf('.') + 1);
+        this.shortName = plugin.alias() == null ? cls.getSimpleName() : plugin.alias().substring(plugin.alias().lastIndexOf('.') + 1);
 
         // outputs
-        this.outputsSchema = jsonSchemaGenerator.outputs(baseCls, cls);
+        this.outputsSchema = jsonSchemaGenerator.outputs(allProperties ? null : plugin.baseClass(), cls);
 
         if (this.outputsSchema.containsKey("$defs")) {
             this.defs.putAll((Map<String, Object>) this.outputsSchema.get("$defs"));
@@ -67,17 +68,13 @@ public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
                 .toList();
         }
 
-        if (alias != null) {
+        if (plugin.alias() != null) {
             this.deprecated = true;
         }
     }
 
-    public static <T> ClassPluginDocumentation<T> of(JsonSchemaGenerator jsonSchemaGenerator, RegisteredPlugin plugin, Class<? extends T> cls, Class<T> baseCls) {
-        return new ClassPluginDocumentation<>(jsonSchemaGenerator, plugin, cls, baseCls, null);
-    }
-
-    public static <T> ClassPluginDocumentation<T> of(JsonSchemaGenerator jsonSchemaGenerator, RegisteredPlugin plugin, Class<? extends T> cls, Class<T> baseCls, String alias) {
-        return new ClassPluginDocumentation<>(jsonSchemaGenerator, plugin, cls, baseCls, alias);
+    public static <T> ClassPluginDocumentation<T> of(JsonSchemaGenerator jsonSchemaGenerator, PluginClassAndMetadata<T> plugin, boolean allProperties) {
+        return new ClassPluginDocumentation<>(jsonSchemaGenerator, plugin, allProperties);
     }
 
     @AllArgsConstructor

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

@@ -7,6 +7,7 @@ import io.kestra.core.models.tasks.logs.LogExporter;
 import io.kestra.core.models.tasks.runners.TaskRunner;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.plugins.PluginClassAndMetadata;
 import io.kestra.core.plugins.RegisteredPlugin;
 import io.kestra.core.runners.pebble.Extension;
 import io.kestra.core.runners.pebble.JsonWriter;
@@ -217,7 +218,15 @@ public class DocumentationGenerator {
     private <T> List<Document> generate(RegisteredPlugin registeredPlugin, List<Class<? extends T>> cls, Class<T> baseCls, String type) {
         return cls
             .stream()
-            .map(r -> ClassPluginDocumentation.of(jsonSchemaGenerator, registeredPlugin, r, baseCls))
+            .map(pluginClass -> {
+                PluginClassAndMetadata<T> metadata = PluginClassAndMetadata.create(
+                    registeredPlugin,
+                    pluginClass,
+                    baseCls,
+                    null
+                );
+                return ClassPluginDocumentation.of(jsonSchemaGenerator, metadata, true);
+            })
             .map(pluginDocumentation -> {
                 try {
                     return new Document(
@@ -247,15 +256,15 @@ public class DocumentationGenerator {
             classPluginDocumentation.getCls() + ".md";
     }
 
-    public static <T> String render(ClassPluginDocumentation<T> classPluginDocumentation) throws IOException {
+    public static String render(ClassPluginDocumentation<?> classPluginDocumentation) throws IOException {
         return render("task", JacksonMapper.toMap(classPluginDocumentation));
     }
 
-    public static <T> String render(AbstractClassDocumentation<T> classInputDocumentation) throws IOException {
+    public static String render(AbstractClassDocumentation classInputDocumentation) throws IOException {
         return render("task", JacksonMapper.toMap(classInputDocumentation));
     }
 
-    public static <T> String render(String templateName, Map<String, Object> vars) throws IOException {
+    public static String render(String templateName, Map<String, Object> vars) throws IOException {
         String pebbleTemplate = IOUtils.toString(
             Objects.requireNonNull(DocumentationGenerator.class.getClassLoader().getResourceAsStream("docs/" + templateName + ".peb")),
             StandardCharsets.UTF_8

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

@@ -0,0 +1,65 @@
+package io.kestra.core.docs;
+
+import io.kestra.core.models.dashboards.Dashboard;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.PluginDefault;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.templates.Template;
+import io.kestra.core.models.triggers.AbstractTrigger;
+import jakarta.inject.Singleton;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Service for getting schemas.
+ */
+@Singleton
+public class JsonSchemaCache {
+
+    private final JsonSchemaGenerator jsonSchemaGenerator;
+
+    private final ConcurrentMap<CacheKey, Map<String, Object>> schemaCache = new ConcurrentHashMap<>();
+
+    private final Map<SchemaType, Class<?>> classesBySchemaType = new HashMap<>();
+
+    /**
+     * Creates a new {@link JsonSchemaCache} instance.
+     *
+     * @param jsonSchemaGenerator The {@link JsonSchemaGenerator}.
+     */
+    public JsonSchemaCache(final JsonSchemaGenerator jsonSchemaGenerator) {
+        this.jsonSchemaGenerator = Objects.requireNonNull(jsonSchemaGenerator, "JsonSchemaGenerator cannot be null");
+        registerClassForType(SchemaType.FLOW, Flow.class);
+        registerClassForType(SchemaType.TEMPLATE, Template.class);
+        registerClassForType(SchemaType.TASK, Task.class);
+        registerClassForType(SchemaType.TRIGGER, AbstractTrigger.class);
+        registerClassForType(SchemaType.PLUGINDEFAULT, PluginDefault.class);
+        registerClassForType(SchemaType.DASHBOARD, Dashboard.class);
+    }
+
+    public Map<String, Object> getSchemaForType(final SchemaType type,
+                                                final boolean arrayOf) {
+        return schemaCache.computeIfAbsent(new CacheKey(type, arrayOf), (key) -> {
+
+            Class<?> cls = Optional.ofNullable(classesBySchemaType.get(type))
+                .orElseThrow(() -> new IllegalArgumentException("Cannot found schema for type '" + type + "'"));
+            return jsonSchemaGenerator.schemas(cls, arrayOf);
+        });
+    }
+
+    public void registerClassForType(final SchemaType type, final Class<?> clazz) {
+        classesBySchemaType.put(type, clazz);
+    }
+
+    public void clear() {
+        schemaCache.clear();
+    }
+
+    private record CacheKey(SchemaType type, boolean arrayOf) {
+    }
+}

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

@@ -64,13 +64,21 @@ public class JsonSchemaGenerator {
         return this.schemas(cls, false);
     }
 
+    private void replaceOneOfWithAnyOf(ObjectNode objectNode) {
+        objectNode.findParents("oneOf").forEach(jsonNode -> {
+            if (jsonNode instanceof ObjectNode oNode) {
+                oNode.set("anyOf", oNode.remove("oneOf"));
+            }
+        });
+    }
+
     public <T> Map<String, Object> schemas(Class<? extends T> cls, boolean arrayOf) {
         SchemaGeneratorConfigBuilder builder = new SchemaGeneratorConfigBuilder(
             SchemaVersion.DRAFT_7,
             OptionPreset.PLAIN_JSON
         );
 
-        this.build(builder,true);
+        this.build(builder, true);
 
         SchemaGeneratorConfig schemaGeneratorConfig = builder.build();
 
@@ -80,8 +88,9 @@ public class JsonSchemaGenerator {
             if (arrayOf) {
                 objectNode.put("type", "array");
             }
-            replaceAnyOfWithOneOf(objectNode);
-            pullOfDefaultFromOneOf(objectNode);
+            replaceOneOfWithAnyOf(objectNode);
+            pullDocumentationAndDefaultFromAnyOf(objectNode);
+            removeRequiredOnPropsWithDefaults(objectNode);
 
             return JacksonMapper.toMap(objectNode);
         } catch (IllegalArgumentException e) {
@@ -89,33 +98,59 @@ public class JsonSchemaGenerator {
         }
     }
 
-    private void replaceAnyOfWithOneOf(ObjectNode objectNode) {
-        objectNode.findParents("anyOf").forEach(jsonNode -> {
-            if (jsonNode instanceof ObjectNode oNode) {
-                oNode.set("oneOf", oNode.remove("anyOf"));
+    private void removeRequiredOnPropsWithDefaults(ObjectNode objectNode) {
+        objectNode.findParents("required").forEach(jsonNode -> {
+            if (jsonNode instanceof ObjectNode clazzSchema && clazzSchema.get("required") instanceof ArrayNode requiredPropsNode && clazzSchema.get("properties") instanceof ObjectNode properties) {
+                List<String> requiredFieldValues = StreamSupport.stream(requiredPropsNode.spliterator(), false)
+                    .map(JsonNode::asText)
+                    .toList();
+
+                properties.fields().forEachRemaining(e -> {
+                    int indexInRequiredArray = requiredFieldValues.indexOf(e.getKey());
+                    if (indexInRequiredArray != -1 && e.getValue() instanceof ObjectNode valueNode && valueNode.has("default")) {
+                        requiredPropsNode.remove(indexInRequiredArray);
+                    }
+                });
+
+                if (requiredPropsNode.isEmpty()) {
+                    clazzSchema.remove("required");
+                }
             }
         });
     }
 
-    // This hack exists because for Property we generate a oneOf for properties that are not strings.
-    // By default, the 'default' is in each oneOf which Monaco editor didn't take into account.
-    // So, we pull off the 'default' from any of the oneOf to the parent.
-    private void pullOfDefaultFromOneOf(ObjectNode objectNode) {
-        objectNode.findParents("oneOf").forEach(jsonNode -> {
+    // This hack exists because for Property we generate a anyOf for properties that are not strings.
+    // By default, the 'default' is in each anyOf which Monaco editor didn't take into account.
+    // So, we pull off the 'default' from any of the anyOf to the parent.
+    // same thing for documentation fields: 'title', 'description', '$deprecated'
+    private void pullDocumentationAndDefaultFromAnyOf(ObjectNode objectNode) {
+        objectNode.findParents("anyOf").forEach(jsonNode -> {
             if (jsonNode instanceof ObjectNode oNode) {
-                JsonNode oneOf = oNode.get("oneOf");
-                if (oneOf instanceof ArrayNode arrayNode) {
+                JsonNode anyOf = oNode.get("anyOf");
+                if (anyOf instanceof ArrayNode arrayNode) {
                     Iterator<JsonNode> it = arrayNode.elements();
-                    JsonNode defaultNode = null;
-                    while (it.hasNext() && defaultNode == null) {
+                    var nodesToPullUp = new HashMap<String, Optional<JsonNode>>(Map.ofEntries(
+                        Map.entry("default", Optional.empty()),
+                        Map.entry("title", Optional.empty()),
+                        Map.entry("description", Optional.empty()),
+                        Map.entry("$deprecated", Optional.empty())
+                    ));
+                    // find nodes to pull up
+                    while (it.hasNext() && nodesToPullUp.containsValue(Optional.<JsonNode>empty())) {
                         JsonNode next = it.next();
                         if (next instanceof ObjectNode nextAsObj) {
-                            defaultNode = nextAsObj.get("default");
+                            nodesToPullUp.entrySet().stream()
+                                .filter(node -> node.getValue().isEmpty())
+                                .forEach(node -> node
+                                    .setValue(Optional.ofNullable(
+                                        nextAsObj.get(node.getKey())
+                                    )));
                         }
                     }
-                    if (defaultNode != null) {
-                        oNode.set("default", defaultNode);
-                    }
+                    // create nodes on parent
+                    nodesToPullUp.entrySet().stream()
+                        .filter(node -> node.getValue().isPresent())
+                        .forEach(node -> oNode.set(node.getKey(), node.getValue().get()));
                 }
             }
         });
@@ -252,11 +287,11 @@ public class JsonSchemaGenerator {
                 TypeContext context = target.getContext();
                 Class<?> erasedType = javaType.getTypeParameters().getFirst().getErasedType();
 
-                if(String.class.isAssignableFrom(erasedType)) {
+                if (String.class.isAssignableFrom(erasedType)) {
                     return List.of(
                         context.resolve(String.class)
                     );
-                } else if(Object.class.equals(erasedType)) {
+                } else if (Object.class.equals(erasedType)) {
                     return List.of(
                         context.resolve(Object.class)
                     );
@@ -296,6 +331,9 @@ public class JsonSchemaGenerator {
                 if (pluginPropertyAnnotation.beta()) {
                     memberAttributes.put("$beta", true);
                 }
+                if (pluginPropertyAnnotation.internalStorageURI()) {
+                    memberAttributes.put("$internalStorageURI", true);
+                }
             }
 
             Schema schema = member.getAnnotationConsideringFieldAndGetter(Schema.class);
@@ -363,7 +401,7 @@ public class JsonSchemaGenerator {
                 // handle deprecated tasks
                 Schema schema = scope.getType().getErasedType().getAnnotation(Schema.class);
                 Deprecated deprecated = scope.getType().getErasedType().getAnnotation(Deprecated.class);
-                if ((schema != null && schema.deprecated()) || deprecated != null ) {
+                if ((schema != null && schema.deprecated()) || deprecated != null) {
                     collectedTypeAttributes.put("$deprecated", "true");
                 }
             });
@@ -388,7 +426,7 @@ public class JsonSchemaGenerator {
         });
 
         // Subtype resolver for all plugins
-        if(builder.build().getSchemaVersion() != SchemaVersion.DRAFT_2019_09) {
+        if (builder.build().getSchemaVersion() != SchemaVersion.DRAFT_2019_09) {
             builder.forTypesInGeneral()
                 .withSubtypeResolver((declaredType, context) -> {
                     TypeContext typeContext = context.getTypeContext();
@@ -577,7 +615,7 @@ public class JsonSchemaGenerator {
         if (property.has("allOf")) {
             for (Iterator<JsonNode> it = property.get("allOf").elements(); it.hasNext(); ) {
                 JsonNode child = it.next();
-                if(child.has("default")) {
+                if (child.has("default")) {
                     return true;
                 }
             }
@@ -591,7 +629,7 @@ public class JsonSchemaGenerator {
             OptionPreset.PLAIN_JSON
         );
 
-        this.build(builder,false);
+        this.build(builder, false);
 
         // we don't return base properties unless specified with @PluginProperty
         builder
@@ -603,8 +641,9 @@ public class JsonSchemaGenerator {
         SchemaGenerator generator = new SchemaGenerator(schemaGeneratorConfig);
         try {
             ObjectNode objectNode = generator.generateSchema(cls);
-            replaceAnyOfWithOneOf(objectNode);
-            pullOfDefaultFromOneOf(objectNode);
+            replaceOneOfWithAnyOf(objectNode);
+            pullDocumentationAndDefaultFromAnyOf(objectNode);
+            removeRequiredOnPropsWithDefaults(objectNode);
 
             return JacksonMapper.toMap(extractMainRef(objectNode));
         } catch (IllegalArgumentException e) {
@@ -714,7 +753,8 @@ public class JsonSchemaGenerator {
 
             field.setAccessible(true);
             return field.invoke(instance);
-        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | IllegalArgumentException ignored) {
+        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException |
+                 IllegalArgumentException ignored) {
 
         }
 
@@ -723,7 +763,8 @@ public class JsonSchemaGenerator {
 
             field.setAccessible(true);
             return field.invoke(instance);
-        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | IllegalArgumentException ignored) {
+        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException |
+                 IllegalArgumentException ignored) {
 
         }
 

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

@@ -7,6 +7,7 @@ import lombok.Data;
 import lombok.NoArgsConstructor;
 
 import java.util.*;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import static java.util.function.Predicate.not;
@@ -50,9 +51,12 @@ public class Plugin {
         if (subgroup == null) {
             plugin.title = registeredPlugin.title();
         } else {
-            subGroupInfos = registeredPlugin.allClass().stream().filter(c -> c.getName().contains(subgroup)).map(clazz -> clazz.getPackage().getDeclaredAnnotation(PluginSubGroup.class)).toList().getFirst();
-            plugin.title = !subGroupInfos.title().isEmpty() ? subGroupInfos.title() : subgroup.substring(subgroup.lastIndexOf('.') + 1);;
-
+            subGroupInfos = registeredPlugin.allClass().stream()
+                .filter(c -> c.getPackageName().contains(subgroup))
+                .min(Comparator.comparingInt(a -> a.getPackageName().length()))
+                .map(clazz -> clazz.getPackage().getDeclaredAnnotation(PluginSubGroup.class))
+                .orElseThrow();
+            plugin.title = !subGroupInfos.title().isEmpty() ? subGroupInfos.title() : subgroup.substring(subgroup.lastIndexOf('.') + 1);
         }
         plugin.group = registeredPlugin.group();
         plugin.description = subGroupInfos != null && !subGroupInfos.description().isEmpty() ? subGroupInfos.description() : registeredPlugin.description();
@@ -74,27 +78,28 @@ public class Plugin {
         plugin.categories = subGroupInfos != null ?
             Arrays.stream(subGroupInfos.categories()).toList() :
             registeredPlugin
-            .allClass()
-            .stream()
-            .map(clazz -> clazz.getPackage().getDeclaredAnnotation(PluginSubGroup.class))
-            .filter(Objects::nonNull)
-            .flatMap(r -> Arrays.stream(r.categories()))
-            .distinct()
-            .toList();
+                .allClass()
+                .stream()
+                .map(clazz -> clazz.getPackage().getDeclaredAnnotation(PluginSubGroup.class))
+                .filter(Objects::nonNull)
+                .flatMap(r -> Arrays.stream(r.categories()))
+                .distinct()
+                .toList();
 
         plugin.subGroup = subgroup;
 
-        plugin.tasks = filterAndGetClassName(registeredPlugin.getTasks(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.triggers = filterAndGetClassName(registeredPlugin.getTriggers(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.conditions = filterAndGetClassName(registeredPlugin.getConditions(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.storages = filterAndGetClassName(registeredPlugin.getStorages(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.secrets = filterAndGetClassName(registeredPlugin.getSecrets(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.taskRunners = filterAndGetClassName(registeredPlugin.getTaskRunners(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.apps = filterAndGetClassName(registeredPlugin.getApps(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.appBlocks = filterAndGetClassName(registeredPlugin.getAppBlocks(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.charts = filterAndGetClassName(registeredPlugin.getCharts(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.dataFilters = filterAndGetClassName(registeredPlugin.getDataFilters(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
-        plugin.logExporters = filterAndGetClassName(registeredPlugin.getLogExporters(), includeDeprecated).stream().filter(c -> subgroup == null || c.startsWith(subgroup)).toList();
+        Predicate<Class<?>> packagePredicate = c -> subgroup == null || c.getPackageName().equals(subgroup);
+        plugin.tasks = filterAndGetClassName(registeredPlugin.getTasks(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.triggers = filterAndGetClassName(registeredPlugin.getTriggers(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.conditions = filterAndGetClassName(registeredPlugin.getConditions(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.storages = filterAndGetClassName(registeredPlugin.getStorages(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.secrets = filterAndGetClassName(registeredPlugin.getSecrets(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.taskRunners = filterAndGetClassName(registeredPlugin.getTaskRunners(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.apps = filterAndGetClassName(registeredPlugin.getApps(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.appBlocks = filterAndGetClassName(registeredPlugin.getAppBlocks(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.charts = filterAndGetClassName(registeredPlugin.getCharts(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.dataFilters = filterAndGetClassName(registeredPlugin.getDataFilters(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.logExporters = filterAndGetClassName(registeredPlugin.getLogExporters(), includeDeprecated, packagePredicate).stream().toList();
 
         return plugin;
     }
@@ -103,15 +108,16 @@ public class Plugin {
      * Filters the given list of class all internal Plugin, as well as, all legacy org.kestra classes.
      * Those classes are only filtered from the documentation to ensure backward compatibility.
      *
-     * @param list The list of classes?
+     * @param list              The list of classes?
      * @param includeDeprecated whether to include deprecated plugins or not
-     * @return  a filtered streams.
+     * @return a filtered streams.
      */
-    private static List<String> filterAndGetClassName(final List<? extends Class<?>> list, boolean includeDeprecated) {
+    private static List<String> filterAndGetClassName(final List<? extends Class<?>> list, boolean includeDeprecated, Predicate<Class<?>> clazzFilter) {
         return list
             .stream()
             .filter(not(io.kestra.core.models.Plugin::isInternal))
             .filter(p -> includeDeprecated || !io.kestra.core.models.Plugin.isDeprecated(p))
+            .filter(clazzFilter)
             .map(Class::getName)
             .filter(c -> !c.startsWith("org.kestra."))
             .toList();

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

@@ -1,11 +1,20 @@
 package io.kestra.core.docs;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import io.kestra.core.utils.Enums;
+
+
 public enum SchemaType {
-    flow,
-    template,
-    task,
-    trigger,
-    plugindefault,
-    apps,
-    dashboard
+    FLOW,
+    TEMPLATE,
+    TASK,
+    TRIGGER,
+    PLUGINDEFAULT,
+    APPS,
+    DASHBOARD;
+
+    @JsonCreator
+    public static SchemaType fromString(final String value) {
+        return Enums.getForNameIgnoreCase(value, SchemaType.class);
+    }
 }

core/src/main/java/io/kestra/core/encryption/EncryptionService.java

@@ -64,8 +64,10 @@ public class EncryptionService {
      * The IV is recovered from the beginning of the string.
      *
      * @see #decrypt(String, byte[])
+     * @throws IllegalArgumentException when the cipherText cannot be BASE64 decoded.
+     *         This may indicate that the cipherText was not encrypted at first so a caller may use this as an indication as it tries to decode a text that was not encoded.
      */
-    public static String decrypt(String key, String cipherText) throws GeneralSecurityException {
+    public static String decrypt(String key, String cipherText) throws GeneralSecurityException, IllegalArgumentException {
         if (cipherText == null || cipherText.isEmpty()) {
             return cipherText;
         }

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

@@ -7,6 +7,8 @@ public enum CrudEventType {
     DELETE,
     LOGIN,
     LOGOUT,
-    IMPERSONATE
+    IMPERSONATE,
+    LOGIN_FAILURE,
+    ACCOUNT_LOCKED
 }
 

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

@@ -23,4 +23,5 @@ public class KestraRuntimeException extends RuntimeException {
     public KestraRuntimeException(Throwable cause) {
         super(cause);
     }
+
 }

core/src/main/java/io/kestra/core/http/HttpRequest.java

@@ -148,30 +148,41 @@ public class HttpRequest {
 
         public abstract String getContentType() throws IOException;
 
+        protected ContentType entityContentType() throws IOException {
+            return this.getCharset() != null ? ContentType.create(this.getContentType(), this.getCharset()) : ContentType.create(this.getContentType());
+        }
+
         public static RequestBody from(HttpEntity entity) throws IOException {
             if (entity == null) {
                 return null;
             }
 
-            Charset charset = entity.getContentEncoding() != null ? Charset.forName(entity.getContentEncoding()) : StandardCharsets.UTF_8;
-
-            if (entity.getContentType().equals(ContentType.APPLICATION_OCTET_STREAM.getMimeType())) {
+            String[] parts = entity.getContentType().split(";");
+            String mimeType = parts[0];
+            Charset charset = StandardCharsets.UTF_8;
+            for (String part : parts) {
+                String stripped = part.strip();
+                if (stripped.startsWith("charset")) {
+                    charset = Charset.forName(stripped.substring(stripped.lastIndexOf('=') + 1));
+                }
+            }
+            if (mimeType.equals(ContentType.APPLICATION_OCTET_STREAM.getMimeType())) {
                 return ByteArrayRequestBody.builder()
-                    .contentType(entity.getContentType())
+                    .contentType(mimeType)
                     .charset(charset)
                     .content(IOUtils.toByteArray(entity.getContent()))
                     .build();
             }
 
-            if (entity.getContentType().equals(ContentType.TEXT_PLAIN.getMimeType())) {
+            if (mimeType.equals(ContentType.TEXT_PLAIN.getMimeType())) {
                 return StringRequestBody.builder()
-                    .contentType(entity.getContentType())
+                    .contentType(mimeType)
                     .charset(charset)
                     .content(IOUtils.toString(entity.getContent(), charset))
                     .build();
             }
 
-            if (entity.getContentType().equals(ContentType.APPLICATION_JSON.getMimeType())) {
+            if (mimeType.equals(ContentType.APPLICATION_JSON.getMimeType())) {
                 return JsonRequestBody.builder()
                     .charset(charset)
                     .content(JacksonMapper.toObject(IOUtils.toString(entity.getContent(), charset)))
@@ -180,7 +191,7 @@ public class HttpRequest {
 
             return ByteArrayRequestBody.builder()
                 .charset(charset)
-                .contentType(entity.getContentType())
+                .contentType(mimeType)
                 .content(entity.getContent().readAllBytes())
                 .build();
         }
@@ -193,13 +204,12 @@ public class HttpRequest {
         @Builder.Default
         private String contentType = ContentType.APPLICATION_OCTET_STREAM.getMimeType();
 
-        @Builder.Default
-        private Charset charset = StandardCharsets.UTF_8;
+        private Charset charset;
 
         private InputStream content;
 
-        public HttpEntity to() {
-            return new InputStreamEntity(content, ContentType.create(contentType, charset));
+        public HttpEntity to() throws IOException {
+            return new InputStreamEntity(content, this.entityContentType());
         }
     }
 
@@ -210,13 +220,12 @@ public class HttpRequest {
         @Builder.Default
         private String contentType = ContentType.TEXT_PLAIN.getMimeType();
 
-        @Builder.Default
-        private Charset charset = StandardCharsets.UTF_8;
+        private Charset charset;
 
         private String content;
 
-        public HttpEntity to() {
-            return new StringEntity(this.content, ContentType.create(contentType, charset));
+        public HttpEntity to() throws IOException {
+            return new StringEntity(this.content, this.entityContentType());
         }
     }
 
@@ -227,13 +236,12 @@ public class HttpRequest {
         @Builder.Default
         private String contentType = ContentType.APPLICATION_OCTET_STREAM.getMimeType();
 
-        @Builder.Default
-        private Charset charset = StandardCharsets.UTF_8;
+        private Charset charset;
 
         private byte[] content;
 
-        public HttpEntity to() {
-            return new ByteArrayEntity(content, ContentType.create(contentType, charset));
+        public HttpEntity to() throws IOException {
+            return new ByteArrayEntity(content, this.entityContentType());
         }
     }
 
@@ -241,8 +249,7 @@ public class HttpRequest {
     @AllArgsConstructor
     @SuperBuilder
     public static class JsonRequestBody extends RequestBody {
-        @Builder.Default
-        private Charset charset = StandardCharsets.UTF_8;
+        private Charset charset;
 
         private Object content;
 
@@ -255,7 +262,7 @@ public class HttpRequest {
             try {
                 return new StringEntity(
                     JacksonMapper.ofJson().writeValueAsString(content),
-                    ContentType.APPLICATION_JSON.withCharset(this.charset)
+                    this.charset != null ? ContentType.APPLICATION_JSON.withCharset(this.charset) : ContentType.APPLICATION_JSON
                 );
             } catch (JsonProcessingException e) {
                 throw new IOException(e);
@@ -267,8 +274,7 @@ public class HttpRequest {
     @AllArgsConstructor
     @SuperBuilder
     public static class UrlEncodedRequestBody extends RequestBody {
-        @Builder.Default
-        private Charset charset = StandardCharsets.UTF_8;
+        private Charset charset;
 
         private Map<String, Object> content;
 
@@ -278,13 +284,12 @@ public class HttpRequest {
         }
 
         public HttpEntity to() throws IOException {
-            return new UrlEncodedFormEntity(
-                this.content .entrySet()
-                    .stream()
-                    .map(e -> new BasicNameValuePair(e.getKey(), e.getValue().toString()))
-                    .toList(),
-                this.charset
-            );
+            List<BasicNameValuePair> list = this.content.entrySet()
+                .stream()
+                .map(e -> new BasicNameValuePair(e.getKey(), e.getValue().toString()))
+                .toList();
+
+            return this.charset != null ? new UrlEncodedFormEntity(list, this.charset) : new UrlEncodedFormEntity(list);
         }
     }
 
@@ -292,8 +297,7 @@ public class HttpRequest {
     @AllArgsConstructor
     @SuperBuilder
     public static class MultipartRequestBody extends RequestBody {
-        @Builder.Default
-        private Charset charset = StandardCharsets.UTF_8;
+        private Charset charset;
 
         private Map<String, Object> content;
 
@@ -304,8 +308,11 @@ public class HttpRequest {
 
         public HttpEntity to() throws IOException {
             MultipartEntityBuilder builder = MultipartEntityBuilder
-                .create()
-                .setCharset(this.charset);
+                .create();
+
+            if (this.charset != null) {
+                builder.setCharset(this.charset);
+            }
 
             content.forEach((key, value) -> {
                 switch (value) {

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

@@ -29,6 +29,7 @@ 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;
@@ -87,47 +88,49 @@ public class HttpClient implements Closeable {
 
         // Timeout
         if (this.configuration.getTimeout() != null) {
-            var connectTiemout = runContext.render(this.configuration.getTimeout().getConnectTimeout()).as(Duration.class);
-            connectTiemout.ifPresent(duration -> connectionConfig.setConnectTimeout(Timeout.of(duration)));
+            var connectTimeout = runContext.render(this.configuration.getTimeout().getConnectTimeout()).as(Duration.class);
+            connectTimeout.ifPresent(duration -> connectionConfig.setConnectTimeout(Timeout.of(duration)));
 
-            var readIdleTiemout = runContext.render(this.configuration.getTimeout().getReadIdleTimeout()).as(Duration.class);
-            readIdleTiemout.ifPresent(duration -> connectionConfig.setSocketTimeout(Timeout.of(duration)));
+            var readIdleTimeout = runContext.render(this.configuration.getTimeout().getReadIdleTimeout()).as(Duration.class);
+            readIdleTimeout.ifPresent(duration -> connectionConfig.setSocketTimeout(Timeout.of(duration)));
         }
 
         // proxy
         if (this.configuration.getProxy() != null && configuration.getProxy().getAddress() != null) {
-            SocketAddress proxyAddr = new InetSocketAddress(
-                runContext.render(configuration.getProxy().getAddress()).as(String.class).orElse(null),
-                runContext.render(configuration.getProxy().getPort()).as(Integer.class).orElse(null)
-            );
+            String proxyAddress = runContext.render(configuration.getProxy().getAddress()).as(String.class).orElse(null);
 
-            Proxy proxy = new Proxy(runContext.render(configuration.getProxy().getType()).as(Proxy.Type.class).orElse(null), proxyAddr);
-
-            builder.setProxySelector(new ProxySelector() {
-                @Override
-                public void connectFailed(URI uri, SocketAddress sa, IOException e) {
-                    /* ignore */
-                }
-
-                @Override
-                public List<Proxy> select(URI uri) {
-                    return List.of(proxy);
-                }
-            });
-
-            if (this.configuration.getProxy().getUsername() != null && this.configuration.getProxy().getPassword() != null) {
-                builder.setProxyAuthenticationStrategy(new DefaultAuthenticationStrategy());
-
-                credentialsStore.setCredentials(
-                    new AuthScope(
-                        runContext.render(this.configuration.getProxy().getAddress()).as(String.class).orElse(null),
-                        runContext.render(this.configuration.getProxy().getPort()).as(Integer.class).orElse(null)
-                    ),
-                    new UsernamePasswordCredentials(
-                        runContext.render(this.configuration.getProxy().getUsername()).as(String.class).orElseThrow(),
-                        runContext.render(this.configuration.getProxy().getPassword()).as(String.class).orElseThrow().toCharArray()
-                    )
+            if (StringUtils.isNotEmpty(proxyAddress)) {
+                int port = runContext.render(configuration.getProxy().getPort()).as(Integer.class).orElseThrow();
+                SocketAddress proxyAddr = new InetSocketAddress(
+                    proxyAddress,
+                    port
                 );
+
+                Proxy proxy = new Proxy(runContext.render(configuration.getProxy().getType()).as(Proxy.Type.class).orElse(null), proxyAddr);
+
+                builder.setProxySelector(new ProxySelector() {
+                    @Override
+                    public void connectFailed(URI uri, SocketAddress sa, IOException e) {
+                        /* ignore */
+                    }
+
+                    @Override
+                    public List<Proxy> select(URI uri) {
+                        return List.of(proxy);
+                    }
+                });
+
+                if (this.configuration.getProxy().getUsername() != null && this.configuration.getProxy().getPassword() != null) {
+                    builder.setProxyAuthenticationStrategy(new DefaultAuthenticationStrategy());
+
+                    credentialsStore.setCredentials(
+                        new AuthScope(proxyAddress, port),
+                        new UsernamePasswordCredentials(
+                            runContext.render(this.configuration.getProxy().getUsername()).as(String.class).orElseThrow(),
+                            runContext.render(this.configuration.getProxy().getPassword()).as(String.class).orElseThrow().toCharArray()
+                        )
+                    );
+                }
             }
         }
 

core/src/main/java/io/kestra/core/http/client/configurations/AbstractAuthConfiguration.java

@@ -3,7 +3,6 @@ package io.kestra.core.http.client.configurations;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
-import io.kestra.core.models.property.Property;
 import io.kestra.core.runners.RunContext;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
@@ -17,7 +16,7 @@ import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
 @SuperBuilder(toBuilder = true)
 @NoArgsConstructor
 public abstract class AbstractAuthConfiguration {
-    public abstract Property<AuthType> getType();
+    public abstract AuthType getType();
 
     public abstract void configure(HttpClientBuilder builder, RunContext runContext) throws IllegalVariableEvaluationException;
 

core/src/main/java/io/kestra/core/http/client/configurations/BasicAuthConfiguration.java

@@ -22,7 +22,7 @@ public class BasicAuthConfiguration extends AbstractAuthConfiguration {
     @NotNull
     @JsonInclude
     @Builder.Default
-    protected Property<AuthType> type = Property.of(AuthType.BASIC);
+    protected AuthType type = AuthType.BASIC;
 
     @Schema(title = "The username for HTTP basic authentication.")
     private Property<String> username;

core/src/main/java/io/kestra/core/http/client/configurations/BearerAuthConfiguration.java

@@ -21,7 +21,7 @@ public class BearerAuthConfiguration extends AbstractAuthConfiguration {
     @NotNull
     @JsonInclude
     @Builder.Default
-    protected Property<AuthType> type = Property.of(AuthType.BEARER);
+    protected AuthType type = AuthType.BEARER;
 
     @Schema(title = "The token for bearer token authentication.")
     private Property<String> token;

core/src/main/java/io/kestra/core/http/client/configurations/HttpConfiguration.java

@@ -2,18 +2,17 @@ package io.kestra.core.http.client.configurations;
 
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.property.Property;
-import io.micronaut.http.client.HttpClientConfiguration;
 import io.micronaut.logging.LogLevel;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Builder;
 import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.jackson.Jacksonized;
 
 import java.net.Proxy;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.time.temporal.ChronoUnit;
 
 @Builder(toBuilder = true)
 @Getter
@@ -30,6 +29,7 @@ public class HttpConfiguration {
     @Schema(title = "The authentification to use.")
     private AbstractAuthConfiguration auth;
 
+    @Setter
     @Schema(title = "The SSL request options")
     private SslOptions ssl;
 
@@ -37,6 +37,7 @@ public class HttpConfiguration {
     @Builder.Default
     private Property<Boolean> followRedirects = Property.of(true);
 
+    @Setter
     @Schema(title = "If true, allow a failed response code (response code >= 400)")
     @Builder.Default
     private Property<Boolean> allowFailed = Property.of(false);
@@ -62,14 +63,12 @@ public class HttpConfiguration {
     private final Duration connectTimeout;
 
     @Schema(title = "The maximum time allowed for reading data from the server before failing.")
-    @Builder.Default
     @Deprecated
-    private final Duration readTimeout = Duration.ofSeconds(HttpClientConfiguration.DEFAULT_READ_TIMEOUT_SECONDS);
+    private final Duration readTimeout;
 
     @Schema(title = "The type of proxy to use.")
-    @Builder.Default
     @Deprecated
-    private final Proxy.Type proxyType = Proxy.Type.DIRECT;
+    private final Proxy.Type proxyType;
 
     @Schema(title = "The address of the proxy server.")
     @Deprecated
@@ -102,19 +101,16 @@ public class HttpConfiguration {
 
     // Deprecated properties with no equivalent value to be kept, silently ignore
     @Schema(title = "The time allowed for a read connection to remain idle before closing it.")
-    @Builder.Default
     @Deprecated
-    private final Duration readIdleTimeout = Duration.of(HttpClientConfiguration.DEFAULT_READ_IDLE_TIMEOUT_MINUTES, ChronoUnit.MINUTES);
+    private final Duration readIdleTimeout;
 
     @Schema(title = "The time an idle connection can remain in the client's connection pool before being closed.")
-    @Builder.Default
     @Deprecated
-    private final Duration connectionPoolIdleTimeout = Duration.ofSeconds(HttpClientConfiguration.DEFAULT_CONNECTION_POOL_IDLE_TIMEOUT_SECONDS);
+    private final Duration connectionPoolIdleTimeout;
 
     @Schema(title = "The maximum content length of the response.")
-    @Builder.Default
     @Deprecated
-    private final Integer maxContentLength = HttpClientConfiguration.DEFAULT_MAX_CONTENT_LENGTH;
+    private final Integer maxContentLength;
 
     public static class HttpConfigurationBuilder {
         @Deprecated

core/src/main/java/io/kestra/core/http/client/configurations/ProxyConfiguration.java

@@ -4,11 +4,13 @@ import io.kestra.core.models.property.Property;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Builder;
 import lombok.Getter;
+import lombok.extern.jackson.Jacksonized;
 
 import java.net.Proxy;
 
 @Getter
 @Builder(toBuilder = true)
+@Jacksonized
 public class ProxyConfiguration {
     @Schema(title = "The type of proxy to use.")
     @Builder.Default

core/src/main/java/io/kestra/core/log/KestraLogFilter.java

@@ -0,0 +1,17 @@
+package io.kestra.core.log;
+
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.boolex.EvaluationException;
+import ch.qos.logback.core.boolex.EventEvaluatorBase;
+
+public class KestraLogFilter extends EventEvaluatorBase<ILoggingEvent> {
+    @Override
+    public boolean evaluate(ILoggingEvent event) throws NullPointerException, EvaluationException {
+        var message = event.getMessage();
+        // as this filter is called very often, for perf,
+        // we use startWith and do all checks successfully instead of using a more elegant construct like Stream...
+        return message.startsWith("outOfOrder mode is active. Migration of schema") ||
+            message.startsWith("Version mismatch         : Database version is older than what dialect POSTGRES supports") ||
+            message.startsWith("Failed to bind as java.util.concurrent.Executors$AutoShutdownDelegatedExecutorService is unsupported.");
+    }
+}

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

@@ -3,6 +3,7 @@ package io.kestra.core.models;
 import io.kestra.core.utils.MapUtils;
 import jakarta.validation.constraints.NotNull;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -18,6 +19,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     public static final String RESTARTED = SYSTEM_PREFIX + "restarted";
     public static final String REPLAY = SYSTEM_PREFIX + "replay";
     public static final String REPLAYED = SYSTEM_PREFIX + "replayed";
+    public static final String SIMULATED_EXECUTION = SYSTEM_PREFIX + "simulatedExecution";
 
     /**
      * Static helper method for converting a list of labels to a nested map.
@@ -46,4 +48,19 @@ public record Label(@NotNull String key, @NotNull String value) {
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .toList();
     }
+
+    /**
+     * Static helper method for converting a label string to a map.
+     *
+     * @param label The label string.
+     * @return The map of key/value labels.
+     */
+    public static Map<String, String> from(String label) {
+        Map<String, String> map = new HashMap<>();
+        String[] keyValueArray = label.split(":");
+        if (keyValueArray.length == 2) {
+            map.put(keyValueArray[0], keyValueArray[1]);
+        }
+        return map;
+    }
 }

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

@@ -0,0 +1,16 @@
+package io.kestra.core.models;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.Pattern;
+
+/**
+ * Interface that can be implemented by classes supporting plugin versioning.
+ *
+ * @see Plugin
+ */
+public interface PluginVersioning {
+
+    @Pattern(regexp="\\d+\\.\\d+\\.\\d+(-[a-zA-Z0-9-]+)?|([a-zA-Z0-9]+)")
+    @Schema(title = "The version of the plugin to use.")
+    String getVersion();
+}

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

@@ -1,9 +1,15 @@
 package io.kestra.core.models;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
+import io.kestra.core.models.dashboards.filters.*;
 import io.kestra.core.utils.Enums;
 import lombok.Builder;
 
-import java.util.*;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -13,33 +19,63 @@ public record QueryFilter(
     Op operation,
     Object value
 ) {
+
+    @JsonCreator
+    public QueryFilter(
+        @JsonProperty("field") Field field,
+        @JsonProperty("operation") Op operation,
+        @JsonProperty("value") Object value
+    ) {
+        this.field = field;
+        this.operation = operation;
+        this.value = value;
+    }
+
     public enum Op {
-        EQUALS("$eq"),
-        NOT_EQUALS("$ne"),
-        GREATER_THAN("$gte"),
-        LESS_THAN("$lte"),
-        IN("$in"),
-        NOT_IN("$notIn"),
-        STARTS_WITH("$startsWith"),
-        ENDS_WITH("$endsWith"),
-        CONTAINS("$contains"),
-        REGEX("$regex");
+        EQUALS,
+        NOT_EQUALS,
+        GREATER_THAN,
+        LESS_THAN,
+        GREATER_THAN_OR_EQUAL_TO,
+        LESS_THAN_OR_EQUAL_TO,
+        IN,
+        NOT_IN,
+        STARTS_WITH,
+        ENDS_WITH,
+        CONTAINS,
+        REGEX;
+    }
 
-        private static final Map<String, Op> BY_VALUE = Arrays.stream(values())
-            .collect(Collectors.toMap(Op::value, Function.identity()));
 
-        private final String value;
-
-        Op(String value) {
-            this.value = value;
-        }
-
-        public static Op fromString(String value) {
-            return Enums.fromString(value, BY_VALUE, "operation");
-        }
-
-        public String value() {
-            return value;
+    @SuppressWarnings("unchecked")
+    public <T extends Enum<T>> AbstractFilter<T> toDashboardFilterBuilder(T field, Object value) {
+        switch (this.operation) {
+            case EQUALS:
+                return EqualTo.<T>builder().field(field).value(value).build();
+            case NOT_EQUALS:
+                return NotEqualTo.<T>builder().field(field).value(value).build();
+            case GREATER_THAN:
+                return GreaterThan.<T>builder().field(field).value(value).build();
+            case LESS_THAN:
+                return LessThan.<T>builder().field(field).value(value).build();
+            case GREATER_THAN_OR_EQUAL_TO:
+                return GreaterThanOrEqualTo.<T>builder().field(field).value(value).build();
+            case LESS_THAN_OR_EQUAL_TO:
+                return LessThanOrEqualTo.<T>builder().field(field).value(value).build();
+            case IN:
+                return In.<T>builder().field(field).values((List<Object>) value).build();
+            case NOT_IN:
+                return NotIn.<T>builder().field(field).values((List<Object>) value).build();
+            case STARTS_WITH:
+                return StartsWith.<T>builder().field(field).value(value.toString()).build();
+            case ENDS_WITH:
+                return EndsWith.<T>builder().field(field).value(value.toString()).build();
+            case CONTAINS:
+                return Contains.<T>builder().field(field).value(value.toString()).build();
+            case REGEX:
+                return Regex.<T>builder().field(field).value(value.toString()).build();
+            default:
+                throw new IllegalArgumentException("Unsupported operation: " + this.operation);
         }
     }
 
@@ -59,7 +95,7 @@ public record QueryFilter(
         NAMESPACE("namespace") {
             @Override
             public List<Op> supportedOp() {
-                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX);
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX, Op.IN);
             }
         },
         LABELS("labels") {
@@ -147,16 +183,17 @@ public record QueryFilter(
             this.value = value;
         }
 
+        @JsonCreator
         public static Field fromString(String value) {
             return Enums.fromString(value, BY_VALUE, "field");
         }
 
+        @JsonValue
         public String value() {
             return value;
         }
     }
 
-
     public enum Resource {
         FLOW {
             @Override
@@ -239,12 +276,17 @@ public record QueryFilter(
         }
 
         private static Operation toOperation(Op op) {
-            return new Operation(op.name(), op.value());
+            return new Operation(op.name(), op.name());
         }
     }
 
-    public record ResourceField(String name, List<FieldOp> fields) {}
-    public record FieldOp(String name, String value, List<Operation> operations) {}
-    public record Operation(String name, String value) {}
+    public record ResourceField(String name, List<FieldOp> fields) {
+    }
+
+    public record FieldOp(String name, String value, List<Operation> operations) {
+    }
+
+    public record Operation(String name, String value) {
+    }
 
 }

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

@@ -4,6 +4,7 @@ import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.repositories.ServiceInstanceRepositoryInterface;
 import io.kestra.core.server.Service;
 import io.kestra.core.server.ServiceInstance;
+import io.kestra.core.server.ServiceType;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
@@ -60,7 +61,7 @@ public record ServiceUsage(
                                   final Duration interval) {
 
         List<DailyServiceStatistics> statistics = Arrays
-            .stream(Service.ServiceType.values())
+            .stream(ServiceType.values())
             .map(type -> of(from, to, repository, type, interval))
             .toList();
         return new ServiceUsage(statistics);
@@ -69,13 +70,13 @@ public record ServiceUsage(
     private static DailyServiceStatistics of(final Instant from,
                                              final Instant to,
                                              final ServiceInstanceRepositoryInterface repository,
-                                             final Service.ServiceType serviceType,
+                                             final ServiceType serviceType,
                                              final Duration interval) {
         return of(serviceType, interval, repository.findAllInstancesBetween(serviceType, from, to));
     }
 
     @VisibleForTesting
-    static DailyServiceStatistics of(final Service.ServiceType serviceType,
+    static DailyServiceStatistics of(final ServiceType serviceType,
                                      final Duration interval,
                                      final List<ServiceInstance> instances) {
         // Compute the number of running service per time-interval.

core/src/main/java/io/kestra/core/models/conditions/ConditionContext.java

@@ -1,5 +1,6 @@
 package io.kestra.core.models.conditions;
 
+import io.kestra.core.models.flows.FlowInterface;
 import lombok.*;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
@@ -18,7 +19,7 @@ import jakarta.validation.constraints.NotNull;
 @AllArgsConstructor
 public class ConditionContext {
     @NotNull
-    private Flow flow;
+    private FlowInterface flow;
 
     private Execution execution;
 

core/src/main/java/io/kestra/core/models/dashboards/DataFilter.java

@@ -1,13 +1,9 @@
 package io.kestra.core.models.dashboards;
 
-import com.fasterxml.jackson.annotation.JsonSubTypes;
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.kestra.core.models.QueryFilter;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.dashboards.filters.AbstractFilter;
 import io.kestra.core.repositories.QueryBuilderInterface;
-import io.kestra.plugin.core.dashboard.data.Executions;
-import io.kestra.plugin.core.dashboard.data.Logs;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
@@ -17,6 +13,7 @@ import lombok.NoArgsConstructor;
 import lombok.Setter;
 import lombok.experimental.SuperBuilder;
 
+import java.time.ZonedDateTime;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -47,6 +44,6 @@ public abstract class DataFilter<F extends Enum<F>, C extends ColumnDescriptor<F
 
     public abstract Class<? extends QueryBuilderInterface<F>> repositoryClass();
 
-    public abstract void setGlobalFilter(GlobalFilter globalFilter);
+    public abstract void setGlobalFilter(List<QueryFilter> queryFilterList, ZonedDateTime startDate, ZonedDateTime endDate);
 
 }

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

@@ -14,6 +14,7 @@ import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.runners.FlowableUtils;
@@ -122,6 +123,10 @@ public class Execution implements DeletedInterface, TenantInterface {
         return newExecution(flow, null, labels, Optional.empty());
     }
 
+    public List<Label> getLabels() {
+        return Optional.ofNullable(this.labels).orElse(new ArrayList<>());
+    }
+
     /**
      * Factory method for constructing a new {@link Execution} object for the given {@link Flow} and
      * inputs.
@@ -131,8 +136,8 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param labels The Flow labels.
      * @return a new {@link Execution}.
      */
-    public static Execution newExecution(final Flow flow,
-        final BiFunction<Flow, Execution, Map<String, Object>> inputs,
+    public static Execution newExecution(final FlowInterface flow,
+        final BiFunction<FlowInterface, Execution, Map<String, Object>> inputs,
         final List<Label> labels,
         final Optional<ZonedDateTime> scheduleDate) {
         Execution execution = builder()
@@ -816,11 +821,20 @@ public class Execution implements DeletedInterface, TenantInterface {
         ));
 
         Map<String, Object> result = new HashMap<>();
-        for (TaskRun current : this.taskRunList) {
-            if (current.getOutputs() != null) {
-                result = MapUtils.merge(result, outputs(current, byIds));
-            }
-        }
+        this.taskRunList.stream()
+            .filter(taskRun -> taskRun.getOutputs() != null)
+            .collect(Collectors.groupingBy(taskRun -> taskRun.getTaskId()))
+            .forEach((taskId, taskRuns) -> {
+                Map<String, Object> taskOutputs = new HashMap<>();
+                for (TaskRun current : taskRuns) {
+                    if (current.getIteration() != null) {
+                        taskOutputs = MapUtils.merge(taskOutputs, outputs(current, byIds));
+                    } else {
+                        taskOutputs.putAll(outputs(current, byIds));
+                    }
+                }
+                result.put(taskId, taskOutputs);
+            });
 
         return result;
     }
@@ -833,18 +847,17 @@ public class Execution implements DeletedInterface, TenantInterface {
 
         if (parents.isEmpty()) {
             if (taskRun.getValue() == null) {
-                return Map.of(taskRun.getTaskId(), taskRun.getOutputs());
+                return taskRun.getOutputs();
             } else {
-                return Map.of(taskRun.getTaskId(),
-                    Map.of(taskRun.getValue(), taskRun.getOutputs()));
+                return Map.of(taskRun.getValue(), taskRun.getOutputs());
             }
         }
 
-        Map<String, Object> result = MapUtils.newHashMap(1);
+        Map<String, Object> result = HashMap.newHashMap(1);
         Map<String, Object> current = result;
 
         for (TaskRun t : parents) {
-            HashMap<String, Object> item = MapUtils.newHashMap(1);
+            HashMap<String, Object> item = HashMap.newHashMap(1);
             current.put(t.getValue(), item);
             current = item;
         }
@@ -857,7 +870,7 @@ public class Execution implements DeletedInterface, TenantInterface {
             }
         }
 
-        return Map.of(taskRun.getTaskId(), result);
+        return result;
     }
 
 

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

@@ -1,8 +1,12 @@
 package io.kestra.core.models.flows;
 
-import io.kestra.core.models.DeletedInterface;
-import io.kestra.core.models.TenantInterface;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.kestra.core.models.Label;
+import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
+import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import io.swagger.v3.oas.annotations.Hidden;
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.*;
 import lombok.Builder;
@@ -11,11 +15,13 @@ import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
 import java.util.List;
+import java.util.Map;
 
 @SuperBuilder(toBuilder = true)
 @Getter
 @NoArgsConstructor
-public abstract class AbstractFlow implements DeletedInterface, TenantInterface {
+@JsonDeserialize
+public abstract class AbstractFlow implements FlowInterface {
     @NotNull
     @NotBlank
     @Pattern(regexp = "^[a-zA-Z0-9][a-zA-Z0-9._-]*")
@@ -33,6 +39,9 @@ public abstract class AbstractFlow implements DeletedInterface, TenantInterface
     @Valid
     List<Input<?>> inputs;
 
+    @Valid
+    List<Output> outputs;
+
     @NotNull
     @Builder.Default
     boolean disabled = false;
@@ -46,4 +55,11 @@ public abstract class AbstractFlow implements DeletedInterface, TenantInterface
     @Pattern(regexp = "^[a-z0-9][a-z0-9_-]*")
     String tenantId;
 
+    @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
+    @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
+    @Schema(implementation = Object.class, oneOf = {List.class, Map.class})
+    List<Label> labels;
+
+    Map<String, Object> variables;
+
 }

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

@@ -6,28 +6,21 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.fasterxml.jackson.databind.introspect.AnnotatedMember;
 import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.HasUID;
-import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.PluginProperty;
-import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.sla.SLA;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.models.triggers.AbstractTrigger;
-import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.serializers.JacksonMapper;
-import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
-import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import io.kestra.core.services.FlowService;
-import io.kestra.core.utils.IdUtils;
+import io.kestra.core.utils.ListUtils;
 import io.kestra.core.validations.FlowValidation;
 import io.micronaut.core.annotation.Introspected;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -44,6 +37,11 @@ import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+/**
+ * A serializable flow with no source.
+ * <p>
+ * This class is planned for deprecation - use the {@link FlowWithSource}.
+ */
 @SuperBuilder(toBuilder = true)
 @Getter
 @NoArgsConstructor
@@ -68,11 +66,6 @@ public class Flow extends AbstractFlow implements HasUID {
 
     String description;
 
-    @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
-    @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
-    @Schema(implementation = Object.class, oneOf = {List.class, Map.class})
-    List<Label> labels;
-
     Map<String, Object> variables;
 
     @Valid
@@ -95,6 +88,9 @@ public class Flow extends AbstractFlow implements HasUID {
     @Deprecated
     List<Listener> listeners;
 
+    @Valid
+    List<Task> afterExecution;
+
     @Valid
     List<AbstractTrigger> triggers;
 
@@ -133,66 +129,6 @@ public class Flow extends AbstractFlow implements HasUID {
     @PluginProperty(beta = true)
     List<SLA> sla;
 
-
-    public Logger logger() {
-        return LoggerFactory.getLogger("flow." + this.id);
-    }
-
-
-    /** {@inheritDoc **/
-    @Override
-    @JsonIgnore
-    public String uid() {
-        return Flow.uid(this.getTenantId(), this.getNamespace(), this.getId(), Optional.ofNullable(this.revision));
-    }
-
-    @JsonIgnore
-    public String uidWithoutRevision() {
-        return Flow.uidWithoutRevision(this.getTenantId(), this.getNamespace(), this.getId());
-    }
-
-    public static String uid(Execution execution) {
-        return IdUtils.fromParts(
-            execution.getTenantId(),
-            execution.getNamespace(),
-            execution.getFlowId(),
-            String.valueOf(execution.getFlowRevision())
-        );
-    }
-
-    public static String uid(String tenantId, String namespace, String id, Optional<Integer> revision) {
-        return IdUtils.fromParts(
-            tenantId,
-            namespace,
-            id,
-            String.valueOf(revision.orElse(-1))
-        );
-    }
-
-    public static String uidWithoutRevision(String tenantId, String namespace, String id) {
-        return IdUtils.fromParts(
-            tenantId,
-            namespace,
-            id
-        );
-    }
-
-    public static String uid(Trigger trigger) {
-        return IdUtils.fromParts(
-            trigger.getTenantId(),
-            trigger.getNamespace(),
-            trigger.getFlowId()
-        );
-    }
-
-    public static String uidWithoutRevision(Execution execution) {
-        return IdUtils.fromParts(
-            execution.getTenantId(),
-            execution.getNamespace(),
-            execution.getFlowId()
-        );
-    }
-
     public Stream<String> allTypes() {
         return Stream.of(
                 Optional.ofNullable(triggers).orElse(Collections.emptyList()).stream().map(AbstractTrigger::getType),
@@ -204,10 +140,10 @@ public class Flow extends AbstractFlow implements HasUID {
 
     public Stream<Task> allTasks() {
         return Stream.of(
-                this.tasks != null ? this.tasks : new ArrayList<Task>(),
-                this.errors != null ? this.errors : new ArrayList<Task>(),
-                this._finally != null ? this._finally : new ArrayList<Task>(),
-                this.listenersTasks()
+                this.tasks != null ? this.tasks : Collections.<Task>emptyList(),
+                this.errors != null ? this.errors : Collections.<Task>emptyList(),
+                this._finally != null ? this._finally : Collections.<Task>emptyList(),
+                this.afterExecutionTasks()
             )
             .flatMap(Collection::stream);
     }
@@ -287,6 +223,14 @@ public class Flow extends AbstractFlow implements HasUID {
             .orElse(null);
     }
 
+    public AbstractTrigger findTriggerByTriggerId(String triggerId) {
+        return this.triggers
+            .stream()
+            .filter(trigger -> trigger.getId().equals(triggerId))
+            .findFirst()
+            .orElse(null);
+    }
+
     /**
      * @deprecated should not be used
      */
@@ -329,18 +273,14 @@ public class Flow extends AbstractFlow implements HasUID {
         }
     }
 
-    private List<Task> listenersTasks() {
-        if (this.getListeners() == null) {
-            return new ArrayList<>();
-        }
-
-        return this.getListeners()
-            .stream()
-            .flatMap(listener -> listener.getTasks().stream())
-            .toList();
+    private List<Task> afterExecutionTasks() {
+        return ListUtils.concat(
+            ListUtils.emptyOnNull(this.getListeners()).stream().flatMap(listener -> listener.getTasks().stream()).toList(),
+            this.getAfterExecution()
+        );
     }
 
-    public boolean equalsWithoutRevision(Flow o) {
+    public boolean equalsWithoutRevision(FlowInterface o) {
         try {
             return WITHOUT_REVISION_OBJECT_MAPPER.writeValueAsString(this).equals(WITHOUT_REVISION_OBJECT_MAPPER.writeValueAsString(o));
         } catch (JsonProcessingException e) {
@@ -380,14 +320,6 @@ public class Flow extends AbstractFlow implements HasUID {
         }
     }
 
-    /**
-     * Convenience method to generate the source of a flow.
-     * Equivalent to <code>FlowService.generateSource(this);</code>
-     */
-    public String generateSource() {
-        return FlowService.generateSource(this);
-    }
-
     public Flow toDeleted() {
         return this.toBuilder()
             .revision(this.revision + 1)
@@ -395,7 +327,13 @@ public class Flow extends AbstractFlow implements HasUID {
             .build();
     }
 
-    public FlowWithSource withSource(String source) {
-        return FlowWithSource.of(this, source);
+    /**
+     * {@inheritDoc}
+     * To be conservative a flow MUST not return any source.
+     */
+    @Override
+    @JsonIgnore
+    public String getSource() {
+        return null;
     }
 }

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

@@ -1,5 +1,6 @@
 package io.kestra.core.models.flows;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.kestra.core.models.tasks.TaskForExecution;
 import io.kestra.core.models.triggers.AbstractTriggerForExecution;
@@ -29,6 +30,9 @@ public class FlowForExecution extends AbstractFlow {
     @JsonProperty("finally")
     List<TaskForExecution> _finally;
 
+    @Valid
+    List<TaskForExecution> afterExecution;
+
     @Valid
     List<AbstractTriggerForExecution> triggers;
 
@@ -42,9 +46,16 @@ public class FlowForExecution extends AbstractFlow {
             .tasks(flow.getTasks().stream().map(TaskForExecution::of).toList())
             .errors(ListUtils.emptyOnNull(flow.getErrors()).stream().map(TaskForExecution::of).toList())
             ._finally(ListUtils.emptyOnNull(flow.getFinally()).stream().map(TaskForExecution::of).toList())
+            .afterExecution(ListUtils.emptyOnNull(flow.getAfterExecution()).stream().map(TaskForExecution::of).toList())
             .triggers(ListUtils.emptyOnNull(flow.getTriggers()).stream().map(AbstractTriggerForExecution::of).toList())
             .disabled(flow.isDisabled())
             .deleted(flow.isDeleted())
             .build();
     }
+
+    @JsonIgnore
+    @Override
+    public String getSource() {
+        return null;
+    }
 }

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

@@ -0,0 +1,71 @@
+package io.kestra.core.models.flows;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.utils.IdUtils;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.util.Optional;
+
+/**
+ * Represents a unique and global identifier for a flow.
+ */
+public interface FlowId {
+
+    String getId();
+
+    String getNamespace();
+
+    Integer getRevision();
+
+    String getTenantId();
+
+
+    static String uid(FlowId flow) {
+        return uid(flow.getTenantId(), flow.getNamespace(), flow.getId(), Optional.ofNullable(flow.getRevision()));
+    }
+
+    static String uid(String tenantId, String namespace, String id, Optional<Integer> revision) {
+        return of(tenantId, namespace, id, revision.orElse(-1)).toString();
+    }
+
+    static String uidWithoutRevision(FlowId flow) {
+        return of(flow.getTenantId(), flow.getNamespace(), flow.getId(), null).toString();
+    }
+
+    static String uidWithoutRevision(String tenantId, String namespace, String id) {
+        return of(tenantId, namespace, id,null).toString();
+    }
+
+    static String uid(Trigger trigger) {
+        return of(trigger.getTenantId(), trigger.getNamespace(), trigger.getFlowId(), null).toString();
+    }
+
+    static String uidWithoutRevision(Execution execution) {
+        return of(execution.getTenantId(), execution.getNamespace(), execution.getFlowId(), null).toString();
+    }
+
+    /**
+     * Static helper method for constructing a new {@link FlowId}.
+     *
+     * @return   a new {@link FlowId}.
+     */
+    static FlowId of(String tenantId, String namespace, String id, Integer revision) {
+        return new Default(tenantId, namespace, id, revision);
+    }
+
+    @Getter
+    @AllArgsConstructor
+    class Default implements FlowId {
+        private final String tenantId;
+        private final String namespace;
+        private final String id;
+        private final Integer revision;
+
+        @Override
+        public String toString() {
+            return IdUtils.fromParts(tenantId, namespace, id, Optional.ofNullable(revision).map(String::valueOf).orElse(null));
+        }
+    }
+}

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

@@ -0,0 +1,194 @@
+package io.kestra.core.models.flows;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.HasSource;
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.Label;
+import io.kestra.core.models.TenantInterface;
+import io.kestra.core.models.flows.sla.SLA;
+import io.kestra.core.serializers.JacksonMapper;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/**
+ * The base interface for FLow.
+ */
+@JsonDeserialize(as = GenericFlow.class)
+public interface FlowInterface extends FlowId, DeletedInterface, TenantInterface, HasUID, HasSource {
+
+    Pattern YAML_REVISION_MATCHER = Pattern.compile("(?m)^revision: \\d+\n?");
+
+    boolean isDisabled();
+
+    boolean isDeleted();
+
+    List<Label> getLabels();
+
+    List<Input<?>> getInputs();
+
+    List<Output> getOutputs();
+
+    Map<String, Object> getVariables();
+
+    default Concurrency getConcurrency() {
+        return null;
+    }
+
+    default List<SLA> getSla() {
+        return List.of();
+    }
+
+    String getSource();
+
+    @Override
+    @JsonIgnore
+    default String source() {
+        return getSource();
+    }
+
+    @Override
+    @JsonIgnore
+    default String uid() {
+        return FlowId.uid(this);
+    }
+
+    @JsonIgnore
+    default String uidWithoutRevision() {
+        return FlowId.uidWithoutRevision(this);
+    }
+
+    /**
+     * Checks whether this flow is equals to the given flow.
+     * <p>
+     * This method is used to compare if two flow revisions are equal.
+     *
+     * @param flow  The flow to compare.
+     * @return {@code true} if both flows are the same. Otherwise {@code false}
+     */
+    @JsonIgnore
+    default boolean isSameWithSource(final FlowInterface flow) {
+        return
+            Objects.equals(this.uidWithoutRevision(), flow.uidWithoutRevision()) &&
+                Objects.equals(this.isDeleted(), flow.isDeleted()) &&
+                Objects.equals(this.isDisabled(), flow.isDisabled()) &&
+                Objects.equals(sourceWithoutRevision(this.getSource()), sourceWithoutRevision(flow.getSource()));
+    }
+
+    /**
+     * Checks whether this flow matches the given {@link FlowId}.
+     *
+     * @param that  The {@link FlowId}.
+     * @return {@code true} if the passed id matches this flow.
+     */
+    @JsonIgnore
+    default boolean isSameId(FlowId that) {
+        if (that == null) return false;
+        return
+            Objects.equals(this.getTenantId(), that.getTenantId()) &&
+            Objects.equals(this.getNamespace(), that.getNamespace()) &&
+            Objects.equals(this.getId(), that.getId());
+    }
+
+    /**
+     * Static method for removing the 'revision' field from a flow.
+     *
+     * @param source    The source.
+     * @return  The source without revision.
+     */
+    static String sourceWithoutRevision(final String source) {
+        return YAML_REVISION_MATCHER.matcher(source).replaceFirst("");
+    }
+
+    /**
+     * Returns the source code for this flow or generate one if {@code null}.
+     * <p>
+     * This method must only be used for testing purpose or for handling backward-compatibility.
+     *
+     * @return the sourcecode.
+     */
+    default String sourceOrGenerateIfNull() {
+        return getSource() != null ? getSource() : SourceGenerator.generate(this);
+    }
+
+    /**
+     * Static helper class for generating source_code from a {@link FlowInterface} object.
+     *
+     * <p>
+     * This class must only be used for testing purpose or for handling backward-compatibility.
+     */
+    class SourceGenerator {
+        private static final ObjectMapper NON_DEFAULT_OBJECT_MAPPER = JacksonMapper.ofJson()
+            .copy()
+            .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
+
+        static String generate(final FlowInterface flow) {
+            try {
+                String json = NON_DEFAULT_OBJECT_MAPPER.writeValueAsString(flow);
+
+                Object map = SourceGenerator.fixSnakeYaml(JacksonMapper.toMap(json));
+
+                String source = JacksonMapper.ofYaml().writeValueAsString(map);
+
+                // remove the revision from the generated source
+                return sourceWithoutRevision(source);
+            } catch (JsonProcessingException e) {
+                return null;
+            }
+        }
+
+        /**
+         * Dirty hack but only concern previous flow with no source code in org.yaml.snakeyaml.emitter.Emitter:
+         * <pre>
+         * if (previousSpace) {
+         *   spaceBreak = true;
+         * }
+         * </pre>
+         * This control will detect ` \n` as a no valid entry on a string and will break the multiline to transform in single line
+         *
+         * @param object the object to fix
+         * @return the modified object
+         */
+        private static Object fixSnakeYaml(Object object) {
+            if (object instanceof Map<?, ?> mapValue) {
+                return mapValue
+                    .entrySet()
+                    .stream()
+                    .map(entry -> new AbstractMap.SimpleEntry<>(
+                        fixSnakeYaml(entry.getKey()),
+                        fixSnakeYaml(entry.getValue())
+                    ))
+                    .filter(entry -> entry.getValue() != null)
+                    .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        Map.Entry::getValue,
+                        (u, v) -> {
+                            throw new IllegalStateException(String.format("Duplicate key %s", u));
+                        },
+                        LinkedHashMap::new
+                    ));
+            } else if (object instanceof Collection<?> collectionValue) {
+                return collectionValue
+                    .stream()
+                    .map(SourceGenerator::fixSnakeYaml)
+                    .toList();
+            } else if (object instanceof String item) {
+                if (item.contains("\n")) {
+                    return item.replaceAll("\\s+\\n", "\\\n");
+                }
+            }
+            return object;
+        }
+    }
+}

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

@@ -1,14 +1,16 @@
 package io.kestra.core.models.flows;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import io.kestra.core.models.executions.Execution;
+import io.kestra.core.serializers.JacksonMapper;
 import io.micronaut.core.annotation.Introspected;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
+import org.slf4j.Logger;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Optional;
 
@@ -21,11 +23,48 @@ import java.util.Optional;
 public class FlowWithException extends FlowWithSource {
     String exception;
 
+    public static FlowWithException from(final FlowInterface flow, final Exception exception) {
+        return FlowWithException.builder()
+            .id(flow.getId())
+            .tenantId(flow.getTenantId())
+            .namespace(flow.getNamespace())
+            .revision(flow.getRevision())
+            .deleted(flow.isDeleted())
+            .exception(exception.getMessage())
+            .tasks(List.of())
+            .source(flow.getSource())
+            .build();
+    }
+
+    public static Optional<FlowWithException> from(final String source, final Exception exception, final Logger log) {
+        log.error("Unable to deserialize a flow: {}", exception.getMessage());
+        try {
+            var jsonNode = JacksonMapper.ofJson().readTree(source);
+            return FlowWithException.from(jsonNode, exception);
+        } catch (IOException e) {
+            // if we cannot create a FlowWithException, ignore the message
+            log.error("Unexpected exception when trying to handle a deserialization error", e);
+            return Optional.empty();
+        }
+    }
+
     public static Optional<FlowWithException> from(JsonNode jsonNode, Exception exception) {
         if (jsonNode.hasNonNull("id") && jsonNode.hasNonNull("namespace")) {
+
+            final String tenantId;
+            if (jsonNode.hasNonNull("tenant_id")) {
+                // JsonNode is from database
+                tenantId = jsonNode.get("tenant_id").asText();
+            } else if (jsonNode.hasNonNull("tenantId")) {
+                // JsonNode is from queue
+                tenantId = jsonNode.get("tenantId").asText();
+            } else {
+                tenantId = null;
+            }
+
             var flow = FlowWithException.builder()
                 .id(jsonNode.get("id").asText())
-                .tenantId(jsonNode.hasNonNull("tenant_id") ? jsonNode.get("tenant_id").asText() : null)
+                .tenantId(tenantId)
                 .namespace(jsonNode.get("namespace").asText())
                 .revision(jsonNode.hasNonNull("revision") ? jsonNode.get("revision").asInt() : 1)
                 .deleted(jsonNode.hasNonNull("deleted") && jsonNode.get("deleted").asBoolean())
@@ -39,4 +78,10 @@ public class FlowWithException extends FlowWithSource {
         // if there is no id and namespace, we return null as we cannot create a meaningful FlowWithException
         return Optional.empty();
     }
+
+    /** {@inheritDoc} **/
+    @Override
+    public Flow toFlow() {
+        return this;
+    }
 }

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

@@ -18,22 +18,14 @@ import lombok.experimental.SuperBuilder;
 @EqualsAndHashCode
 @FlowValidation
 public class FlowWithPath {
-    private FlowWithSource flow;
+    private FlowInterface flow;
     @Nullable
     private String tenantId;
     private String id;
     private String namespace;
     private String path;
 
-    public static FlowWithPath of(FlowWithSource flow, String path) {
-        return FlowWithPath.builder()
-            .id(flow.getId())
-            .namespace(flow.getNamespace())
-            .path(path)
-            .build();
-    }
-
-    public static FlowWithPath of(Flow flow, String path) {
+    public static FlowWithPath of(FlowInterface flow, String path) {
         return FlowWithPath.builder()
             .id(flow.getId())
             .namespace(flow.getNamespace())

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

@@ -1,18 +1,22 @@
 package io.kestra.core.models.flows;
 
-import io.kestra.core.models.HasSource;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.micronaut.core.annotation.Introspected;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
+import java.util.Objects;
+import java.util.regex.Pattern;
+
 @SuperBuilder(toBuilder = true)
 @Getter
 @NoArgsConstructor
 @Introspected
 @ToString
-public class FlowWithSource extends Flow implements HasSource {
+public class FlowWithSource extends Flow {
+
     String source;
 
     @SuppressWarnings("deprecation")
@@ -31,6 +35,7 @@ public class FlowWithSource extends Flow implements HasSource {
             .errors(this.errors)
             ._finally(this._finally)
             .listeners(this.listeners)
+            .afterExecution(this.afterExecution)
             .triggers(this.triggers)
             .pluginDefaults(this.pluginDefaults)
             .disabled(this.disabled)
@@ -41,15 +46,13 @@ public class FlowWithSource extends Flow implements HasSource {
             .build();
     }
 
-    private static String cleanupSource(String source) {
-        return source.replaceFirst("(?m)^revision: \\d+\n?","");
-    }
-
-    public boolean equals(Flow flow, String flowSource) {
-        return this.equalsWithoutRevision(flow) &&
-            this.source.equals(cleanupSource(flowSource));
+    @Override
+    @JsonIgnore(value = false)
+    public String getSource() {
+        return this.source;
     }
 
+    @Override
     public FlowWithSource toDeleted() {
         return this.toBuilder()
             .revision(this.revision + 1)
@@ -72,6 +75,7 @@ public class FlowWithSource extends Flow implements HasSource {
             .tasks(flow.tasks)
             .errors(flow.errors)
             ._finally(flow._finally)
+            .afterExecution(flow.afterExecution)
             .listeners(flow.listeners)
             .triggers(flow.triggers)
             .pluginDefaults(flow.pluginDefaults)
@@ -83,10 +87,4 @@ public class FlowWithSource extends Flow implements HasSource {
             .sla(flow.sla)
             .build();
     }
-
-    /** {@inheritDoc} **/
-    @Override
-    public String source() {
-        return getSource();
-    }
 }